diff options
Diffstat (limited to 'storage/maria/ma_create.c')
| -rw-r--r-- | storage/maria/ma_create.c | 1526 |
1 files changed, 1526 insertions, 0 deletions
diff --git a/storage/maria/ma_create.c b/storage/maria/ma_create.c new file mode 100644 index 00000000..7fd739d1 --- /dev/null +++ b/storage/maria/ma_create.c @@ -0,0 +1,1526 @@ +/* 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 */ + +/* Create a MARIA table */ + +#include "ma_ftdefs.h" +#include "ma_sp_defs.h" +#include <my_bit.h> +#include "ma_blockrec.h" +#include "trnman_public.h" +#include "trnman.h" +#include "ma_crypt.h" + +#ifdef _WIN32 +#include <fcntl.h> +#endif +#include <m_ctype.h> + +static int compare_columns(MARIA_COLUMNDEF **a, MARIA_COLUMNDEF **b); + + +static ulonglong update_tot_length(ulonglong tot_length, ulonglong max_rows, uint length) +{ + ulonglong tot_length_part; + + if (tot_length == ULONGLONG_MAX) + return ULONGLONG_MAX; + + tot_length_part= (max_rows/(ulong) ((maria_block_size - + MAX_KEYPAGE_HEADER_SIZE - KEYPAGE_CHECKSUM_SIZE)/ + (length*2))); + if (tot_length_part >= ULONGLONG_MAX / maria_block_size) + return ULONGLONG_MAX; + + if (tot_length > ULONGLONG_MAX - tot_length_part * maria_block_size) + return ULONGLONG_MAX; + + return tot_length + tot_length_part * maria_block_size; +} + + +/* + Old options is used when recreating database, from maria_chk +*/ + +int maria_create(const char *name, enum data_file_type datafile_type, + uint keys,MARIA_KEYDEF *keydefs, + uint columns, MARIA_COLUMNDEF *columndef, + uint uniques, MARIA_UNIQUEDEF *uniquedefs, + MARIA_CREATE_INFO *ci,uint flags) +{ + uint i,j; + File UNINIT_VAR(dfile), UNINIT_VAR(file); + int errpos,save_errno, create_mode= O_RDWR | O_TRUNC, res; + myf create_flag, common_flag= MY_WME, sync_dir= 0; + uint length,max_key_length,packed,pack_bytes,pointer,real_length_diff, + key_length,info_length,key_segs,options,min_key_length, + base_pos,long_varchar_count, + unique_key_parts,fulltext_keys,offset, not_block_record_extra_length; + uint max_field_lengths, extra_header_size, column_nr; + uint internal_table= flags & HA_CREATE_INTERNAL_TABLE; + ulong reclength, real_reclength,min_pack_length; + char kfilename[FN_REFLEN], klinkname[FN_REFLEN], *klinkname_ptr= 0; + char dfilename[FN_REFLEN], dlinkname[FN_REFLEN], *dlinkname_ptr= 0; + ulong pack_reclength; + ulonglong tot_length,max_rows, tmp; + enum en_fieldtype type; + enum data_file_type org_datafile_type= datafile_type; + MARIA_SHARE share; + TRN tmp_transaction_object; + MARIA_KEYDEF *keydef,tmp_keydef; + MARIA_UNIQUEDEF *uniquedef; + HA_KEYSEG *keyseg,tmp_keyseg; + MARIA_COLUMNDEF *column, *end_column; + double *rec_per_key_part; + ulong *nulls_per_key_part; + uint16 *column_array; + my_off_t key_root[HA_MAX_POSSIBLE_KEY], kfile_size_before_extension; + MARIA_CREATE_INFO tmp_create_info; + my_bool tmp_table= FALSE; /* cache for presence of HA_OPTION_TMP_TABLE */ + my_bool forced_packed; + uchar *log_data= NULL; + my_bool encrypted= ci->encrypted && datafile_type == BLOCK_RECORD; + my_bool insert_order= MY_TEST(flags & HA_PRESERVE_INSERT_ORDER); + uint crypt_page_header_space= 0; + DBUG_ENTER("maria_create"); + DBUG_PRINT("enter", ("keys: %u columns: %u uniques: %u flags: %u", + keys, columns, uniques, flags)); + + DBUG_ASSERT(maria_inited); + + if (flags & HA_CREATE_TMP_TABLE) + common_flag|= MY_THREAD_SPECIFIC; + + if (!ci) + { + bzero((char*) &tmp_create_info,sizeof(tmp_create_info)); + ci=&tmp_create_info; + } + + if (keys + uniques > MARIA_MAX_KEY) + { + DBUG_RETURN(my_errno=HA_WRONG_CREATE_OPTION); + } + errpos=0; + options=0; + bzero((uchar*) &share,sizeof(share)); + + if (flags & HA_DONT_TOUCH_DATA) + { + /* We come here from recreate table */ + org_datafile_type= ci->org_data_file_type; + if (!(ci->old_options & HA_OPTION_TEMP_COMPRESS_RECORD)) + options= (ci->old_options & + (HA_OPTION_COMPRESS_RECORD | HA_OPTION_PACK_RECORD | + HA_OPTION_READ_ONLY_DATA | HA_OPTION_CHECKSUM | + HA_OPTION_TMP_TABLE | HA_OPTION_DELAY_KEY_WRITE | + HA_OPTION_LONG_BLOB_PTR | HA_OPTION_PAGE_CHECKSUM)); + else + { + /* Uncompressing rows */ + options= (ci->old_options & + (HA_OPTION_CHECKSUM | HA_OPTION_TMP_TABLE | + HA_OPTION_DELAY_KEY_WRITE | HA_OPTION_LONG_BLOB_PTR | + HA_OPTION_PAGE_CHECKSUM)); + } + } + else + { + /* Transactional tables must be of type BLOCK_RECORD */ + if (ci->transactional) + datafile_type= BLOCK_RECORD; + } + + if (!(rec_per_key_part= + (double*) my_malloc(PSI_INSTRUMENT_ME, + (keys + uniques)*HA_MAX_KEY_SEG*sizeof(double) + + (keys + uniques)*HA_MAX_KEY_SEG*sizeof(ulong) + + sizeof(uint16) * columns, + MYF(common_flag | MY_ZEROFILL)))) + DBUG_RETURN(my_errno); + nulls_per_key_part= (ulong*) (rec_per_key_part + + (keys + uniques) * HA_MAX_KEY_SEG); + column_array= (uint16*) (nulls_per_key_part + + (keys + uniques) * HA_MAX_KEY_SEG); + + + /* Start by checking fields and field-types used */ + long_varchar_count=packed= not_block_record_extra_length= + pack_reclength= max_field_lengths= 0; + reclength= min_pack_length= ci->null_bytes; + forced_packed= 0; + column_nr= 0; + + if (encrypted) + { + DBUG_ASSERT(datafile_type == BLOCK_RECORD); + crypt_page_header_space= ma_crypt_get_data_page_header_space(); + } + + for (column= columndef, end_column= column + columns ; + column != end_column ; + column++) + { + /* Fill in not used struct parts */ + column->column_nr= column_nr++; + column->offset= reclength; + column->empty_pos= 0; + column->empty_bit= 0; + column->fill_length= column->length; + if (column->null_bit) + options|= HA_OPTION_NULL_FIELDS; + + reclength+= column->length; + type= column->type; + if (datafile_type == BLOCK_RECORD) + { + if (type == FIELD_SKIP_PRESPACE) + type= column->type= FIELD_NORMAL; /* SKIP_PRESPACE not supported */ + if (type == FIELD_NORMAL && + column->length > FULL_PAGE_SIZE2(maria_block_size, + crypt_page_header_space)) + { + /* FIELD_NORMAL can't be split over many blocks, convert to a CHAR */ + type= column->type= FIELD_SKIP_ENDSPACE; + } + } + + if (type != FIELD_NORMAL && type != FIELD_CHECK) + { + column->empty_pos= packed/8; + column->empty_bit= (1 << (packed & 7)); + if (type == FIELD_BLOB) + { + forced_packed= 1; + packed++; + share.base.blobs++; + if (pack_reclength != INT_MAX32) + { + if (column->length == 4+portable_sizeof_char_ptr) + pack_reclength= INT_MAX32; + else + { + /* Add max possible blob length */ + pack_reclength+= (1 << ((column->length- + portable_sizeof_char_ptr)*8)); + } + } + max_field_lengths+= (column->length - portable_sizeof_char_ptr); + } + else if (type == FIELD_SKIP_PRESPACE || + type == FIELD_SKIP_ENDSPACE) + { + forced_packed= 1; + max_field_lengths+= column->length > 255 ? 2 : 1; + not_block_record_extra_length++; + packed++; + } + else if (type == FIELD_VARCHAR) + { + pack_reclength++; + not_block_record_extra_length++; + max_field_lengths++; + if (datafile_type != DYNAMIC_RECORD) + packed++; + column->fill_length= 1; + options|= HA_OPTION_NULL_FIELDS; /* Use ma_checksum() */ + + /* We must test for 257 as length includes pack-length */ + if (MY_TEST(column->length >= 257)) + { + long_varchar_count++; + max_field_lengths++; + column->fill_length= 2; + } + } + else if (type == FIELD_SKIP_ZERO) + packed++; + else + { + if (!column->null_bit) + min_pack_length+= column->length; + else + { + /* Only BLOCK_RECORD skips NULL fields for all field values */ + not_block_record_extra_length+= column->length; + } + column->empty_pos= 0; + column->empty_bit= 0; + } + } + else /* FIELD_NORMAL */ + { + if (!column->null_bit) + { + min_pack_length+= column->length; + share.base.fixed_not_null_fields++; + share.base.fixed_not_null_fields_length+= column->length; + } + else + not_block_record_extra_length+= column->length; + } + } + + if (datafile_type == STATIC_RECORD && forced_packed) + { + /* Can't use fixed length records, revert to block records */ + datafile_type= BLOCK_RECORD; + } + + if (datafile_type == NO_RECORD && uniques) + { + /* Can't do unique without data, revert to block records */ + datafile_type= BLOCK_RECORD; + } + + if (encrypted) + { + /* + datafile_type is set (finally?) + update encryption that is only supported for BLOCK_RECORD + */ + if (datafile_type != BLOCK_RECORD) + { + encrypted= FALSE; + crypt_page_header_space= 0; + } + } + + if (datafile_type == DYNAMIC_RECORD) + options|= HA_OPTION_PACK_RECORD; /* Must use packed records */ + + if (datafile_type == STATIC_RECORD || datafile_type == NO_RECORD) + { + /* We can't use checksum with static length rows */ + flags&= ~HA_CREATE_CHECKSUM; + options&= ~HA_OPTION_CHECKSUM; + min_pack_length= reclength; + packed= 0; + } + else if (datafile_type != BLOCK_RECORD) + min_pack_length+= not_block_record_extra_length; + else + min_pack_length+= 5; /* Min row overhead */ + + if (flags & HA_CREATE_TMP_TABLE) + { + options|= HA_OPTION_TMP_TABLE; + tmp_table= TRUE; + create_mode|= O_NOFOLLOW | (internal_table ? 0 : O_EXCL); + /* "CREATE TEMPORARY" tables are not crash-safe (dropped at restart) */ + ci->transactional= FALSE; + flags&= ~HA_CREATE_PAGE_CHECKSUM; + } + share.base.null_bytes= ci->null_bytes; + share.base.original_null_bytes= ci->null_bytes; + share.base.born_transactional= ci->transactional; + share.base.max_field_lengths= max_field_lengths; + share.base.field_offsets= 0; /* for future */ + share.base.compression_algorithm= ci->compression_algorithm; + share.base.s3_block_size= ci->s3_block_size; + + if (flags & HA_CREATE_CHECKSUM || (options & HA_OPTION_CHECKSUM)) + { + options|= HA_OPTION_CHECKSUM; + min_pack_length++; + pack_reclength++; + } + if (pack_reclength < INT_MAX32) + pack_reclength+= max_field_lengths + long_varchar_count; + else + pack_reclength= INT_MAX32; + + if (flags & HA_CREATE_DELAY_KEY_WRITE) + options|= HA_OPTION_DELAY_KEY_WRITE; + if (flags & HA_CREATE_RELIES_ON_SQL_LAYER) + options|= HA_OPTION_RELIES_ON_SQL_LAYER; + if (flags & HA_CREATE_PAGE_CHECKSUM) + options|= HA_OPTION_PAGE_CHECKSUM; + + pack_bytes= (packed + 7) / 8; + if (pack_reclength != INT_MAX32) + pack_reclength+= reclength+pack_bytes + + MY_TEST(test_all_bits(options, HA_OPTION_CHECKSUM | + HA_OPTION_PACK_RECORD)); + min_pack_length+= pack_bytes; + /* Calculate min possible row length for rows-in-block */ + extra_header_size= MAX_FIXED_HEADER_SIZE; + if (ci->transactional) + { + extra_header_size= TRANS_MAX_FIXED_HEADER_SIZE; + DBUG_PRINT("info",("creating a transactional table")); + } + share.base.min_block_length= (extra_header_size + share.base.null_bytes + + pack_bytes); + if (!ci->data_file_length && ci->max_rows) + { + set_if_bigger(ci->max_rows, ci->reloc_rows); + if (pack_reclength == INT_MAX32 || + (~(ulonglong) 0)/ci->max_rows < (ulonglong) pack_reclength) + ci->data_file_length= ~(ulonglong) 0; + else + { + ci->data_file_length= _ma_safe_mul(ci->max_rows, pack_reclength); + if (datafile_type == BLOCK_RECORD) + { + /* Assume that blocks are only half full (very pessimistic!) */ + ci->data_file_length= _ma_safe_mul(ci->data_file_length, 2); + set_if_bigger(ci->data_file_length, maria_block_size*2); + } + } + } + else if (!ci->max_rows) + { + if (datafile_type == BLOCK_RECORD) + { + uint rows_per_page= + ((maria_block_size - PAGE_OVERHEAD_SIZE_RAW - crypt_page_header_space) + / (min_pack_length + extra_header_size + DIR_ENTRY_SIZE)); + ulonglong data_file_length= ci->data_file_length; + if (!data_file_length) + data_file_length= ((((ulonglong) 1 << ((BLOCK_RECORD_POINTER_SIZE-1) * + 8))/2 -1) * maria_block_size); + if (rows_per_page > 0) + { + set_if_smaller(rows_per_page, MAX_ROWS_PER_PAGE); + ci->max_rows= (data_file_length / maria_block_size+1) * rows_per_page; + } + else + ci->max_rows= data_file_length / (min_pack_length + + extra_header_size + + DIR_ENTRY_SIZE); + } + else + ci->max_rows=(ha_rows) (ci->data_file_length/(min_pack_length + + ((options & + HA_OPTION_PACK_RECORD) ? + 3 : 0))); + set_if_smaller(ci->reloc_rows, ci->max_rows); + } + max_rows= (ulonglong) ci->max_rows; + if (datafile_type == BLOCK_RECORD) + { + /* + The + 1 is for record position withing page + The * 2 is because we need one bit for knowing if there is transid's + after the row pointer + */ + pointer= maria_get_pointer_length((ci->data_file_length / + maria_block_size) * 2, 4) + 1; + set_if_smaller(pointer, BLOCK_RECORD_POINTER_SIZE); + + if (!max_rows) + max_rows= (((((ulonglong) 1 << ((pointer-1)*8)) -1) * maria_block_size) / + min_pack_length / 2); + } + else + { + if (datafile_type == NO_RECORD) + pointer= 0; + else if (datafile_type != STATIC_RECORD) + pointer= maria_get_pointer_length(ci->data_file_length, + maria_data_pointer_size); + else + pointer= maria_get_pointer_length(ci->max_rows, maria_data_pointer_size); + if (!max_rows) + max_rows= ((((ulonglong) 1 << (pointer*8)) -1) / min_pack_length); + } + + real_reclength=reclength; + if (datafile_type == STATIC_RECORD) + { + if (reclength <= pointer) + reclength=pointer+1; /* reserve place for delete link */ + } + else + reclength+= long_varchar_count; /* We need space for varchar! */ + + max_key_length=0; tot_length=0 ; key_segs=0; + fulltext_keys=0; + share.state.rec_per_key_part= rec_per_key_part; + share.state.nulls_per_key_part= nulls_per_key_part; + share.state.key_root=key_root; + share.state.key_del= HA_OFFSET_ERROR; + if (uniques) + max_key_length= MARIA_UNIQUE_HASH_LENGTH + pointer; + + for (i=0, keydef=keydefs ; i < keys ; i++ , keydef++) + { + share.state.key_root[i]= HA_OFFSET_ERROR; + length= real_length_diff= 0; + min_key_length= key_length= pointer; + + if (keydef->key_alg == HA_KEY_ALG_RTREE) + keydef->flag|= HA_RTREE_INDEX; /* For easier tests */ + + if (keydef->flag & HA_SPATIAL) + { +#ifdef HAVE_SPATIAL + /* BAR TODO to support 3D and more dimensions in the future */ + uint sp_segs=SPDIMS*2; + keydef->flag=HA_SPATIAL; + + if (flags & HA_DONT_TOUCH_DATA) + { + /* + Called by maria_chk - i.e. table structure was taken from + MYI file and SPATIAL key *does have* additional sp_segs keysegs. + keydef->seg here points right at the GEOMETRY segment, + so we only need to decrease keydef->keysegs. + (see maria_recreate_table() in _ma_check.c) + */ + keydef->keysegs-=sp_segs-1; + } + + for (j=0, keyseg=keydef->seg ; (int) j < keydef->keysegs ; + j++, keyseg++) + { + if (keyseg->type != HA_KEYTYPE_BINARY && + keyseg->type != HA_KEYTYPE_VARBINARY1 && + keyseg->type != HA_KEYTYPE_VARBINARY2) + { + my_errno=HA_WRONG_CREATE_OPTION; + goto err_no_lock; + } + } + keydef->keysegs+=sp_segs; + key_length+=SPLEN*sp_segs; + length++; /* At least one length uchar */ + min_key_length++; +#else + my_errno= HA_ERR_UNSUPPORTED; + goto err_no_lock; +#endif /*HAVE_SPATIAL*/ + } + else if (keydef->flag & HA_FULLTEXT) + { + keydef->flag=HA_FULLTEXT | HA_PACK_KEY | HA_VAR_LENGTH_KEY; + options|=HA_OPTION_PACK_KEYS; /* Using packed keys */ + + for (j=0, keyseg=keydef->seg ; (int) j < keydef->keysegs ; + j++, keyseg++) + { + if (keyseg->type != HA_KEYTYPE_TEXT && + keyseg->type != HA_KEYTYPE_VARTEXT1 && + keyseg->type != HA_KEYTYPE_VARTEXT2) + { + my_errno=HA_WRONG_CREATE_OPTION; + goto err_no_lock; + } + if (!(keyseg->flag & HA_BLOB_PART) && + (keyseg->type == HA_KEYTYPE_VARTEXT1 || + keyseg->type == HA_KEYTYPE_VARTEXT2)) + { + /* Make a flag that this is a VARCHAR */ + keyseg->flag|= HA_VAR_LENGTH_PART; + /* Store in bit_start number of bytes used to pack the length */ + keyseg->bit_start= ((keyseg->type == HA_KEYTYPE_VARTEXT1)? + 1 : 2); + } + } + + fulltext_keys++; + key_length+= HA_FT_MAXBYTELEN+HA_FT_WLEN; + length++; /* At least one length uchar */ + min_key_length+= 1 + HA_FT_WLEN; + real_length_diff=HA_FT_MAXBYTELEN-FT_MAX_WORD_LEN_FOR_SORT; + } + else + { + /* Test if prefix compression */ + if (keydef->flag & HA_PACK_KEY) + { + /* Can't use space_compression on number keys */ + if ((keydef->seg[0].flag & HA_SPACE_PACK) && + keydef->seg[0].type == (int) HA_KEYTYPE_NUM) + keydef->seg[0].flag&= ~HA_SPACE_PACK; + + /* Only use HA_PACK_KEY when first segment is a variable length key */ + if (!(keydef->seg[0].flag & (HA_SPACE_PACK | HA_BLOB_PART | + HA_VAR_LENGTH_PART))) + { + /* pack relative to previous key */ + keydef->flag&= ~HA_PACK_KEY; + keydef->flag|= HA_BINARY_PACK_KEY | HA_VAR_LENGTH_KEY; + } + else + { + keydef->seg[0].flag|=HA_PACK_KEY; /* for easyer intern test */ + keydef->flag|=HA_VAR_LENGTH_KEY; + options|=HA_OPTION_PACK_KEYS; /* Using packed keys */ + } + } + if (keydef->flag & HA_BINARY_PACK_KEY) + options|=HA_OPTION_PACK_KEYS; /* Using packed keys */ + + if (keydef->flag & HA_AUTO_KEY && ci->with_auto_increment) + share.base.auto_key=i+1; + for (j=0, keyseg=keydef->seg ; j < keydef->keysegs ; j++, keyseg++) + { + /* numbers are stored with high by first to make compression easier */ + switch (keyseg->type) { + case HA_KEYTYPE_SHORT_INT: + case HA_KEYTYPE_LONG_INT: + case HA_KEYTYPE_FLOAT: + case HA_KEYTYPE_DOUBLE: + case HA_KEYTYPE_USHORT_INT: + case HA_KEYTYPE_ULONG_INT: + case HA_KEYTYPE_LONGLONG: + case HA_KEYTYPE_ULONGLONG: + case HA_KEYTYPE_INT24: + case HA_KEYTYPE_UINT24: + case HA_KEYTYPE_INT8: + keyseg->flag|= HA_SWAP_KEY; + break; + case HA_KEYTYPE_VARTEXT1: + case HA_KEYTYPE_VARTEXT2: + case HA_KEYTYPE_VARBINARY1: + case HA_KEYTYPE_VARBINARY2: + if (!(keyseg->flag & HA_BLOB_PART)) + { + /* Make a flag that this is a VARCHAR */ + keyseg->flag|= HA_VAR_LENGTH_PART; + /* Store in bit_start number of bytes used to pack the length */ + keyseg->bit_start= ((keyseg->type == HA_KEYTYPE_VARTEXT1 || + keyseg->type == HA_KEYTYPE_VARBINARY1) ? + 1 : 2); + } + break; + default: + break; + } + if (keyseg->flag & HA_SPACE_PACK) + { + DBUG_ASSERT(!(keyseg->flag & (HA_VAR_LENGTH_PART | HA_BLOB_PART))); + keydef->flag |= HA_SPACE_PACK_USED | HA_VAR_LENGTH_KEY; + options|=HA_OPTION_PACK_KEYS; /* Using packed keys */ + length++; /* At least one length uchar */ + if (!keyseg->null_bit) + min_key_length++; + key_length+= keyseg->length; + if (keyseg->length >= 255) + { + /* prefix may be 3 bytes */ + length+= 2; + } + } + else if (keyseg->flag & (HA_VAR_LENGTH_PART | HA_BLOB_PART)) + { + DBUG_ASSERT(!test_all_bits(keyseg->flag, + (HA_VAR_LENGTH_PART | HA_BLOB_PART))); + keydef->flag|=HA_VAR_LENGTH_KEY; + length++; /* At least one length uchar */ + if (!keyseg->null_bit) + min_key_length++; + options|=HA_OPTION_PACK_KEYS; /* Using packed keys */ + key_length+= keyseg->length; + if (keyseg->length >= 255) + { + /* prefix may be 3 bytes */ + length+= 2; + } + } + else + { + key_length+= keyseg->length; + if (!keyseg->null_bit) + min_key_length+= keyseg->length; + } + if (keyseg->null_bit) + { + key_length++; + /* min key part is 1 byte */ + min_key_length++; + options|=HA_OPTION_PACK_KEYS; + keyseg->flag|=HA_NULL_PART; + keydef->flag|=HA_VAR_LENGTH_KEY | HA_NULL_PART_KEY; + } + } + } /* if HA_FULLTEXT */ + key_segs+=keydef->keysegs; + if (keydef->keysegs > HA_MAX_KEY_SEG) + { + my_errno=HA_WRONG_CREATE_OPTION; + goto err_no_lock; + } + /* + key_segs may be 0 in the case when we only want to be able to + add on row into the table. This can happen with some DISTINCT queries + in MySQL + */ + if ((keydef->flag & (HA_NOSAME | HA_NULL_PART_KEY)) == HA_NOSAME && + key_segs) + share.state.rec_per_key_part[key_segs-1]=1L; + length+=key_length; + /* + A key can't be longer than than half a index block (as we have + to be able to put at least 2 keys on an index block for the key + algorithms to work). + */ + if (length > _ma_max_key_length()) + { + my_errno=HA_WRONG_CREATE_OPTION; + goto err_no_lock; + } + keydef->block_length= (uint16) maria_block_size; + keydef->keylength= (uint16) key_length; + keydef->minlength= (uint16) min_key_length; + keydef->maxlength= (uint16) length; + + if (length > max_key_length) + max_key_length= length; + + tot_length= update_tot_length(tot_length, max_rows, length); + } + + unique_key_parts=0; + for (i=0, uniquedef=uniquedefs ; i < uniques ; i++ , uniquedef++) + { + uniquedef->key=keys+i; + unique_key_parts+=uniquedef->keysegs; + share.state.key_root[keys+i]= HA_OFFSET_ERROR; + + tot_length= update_tot_length(tot_length, max_rows, MARIA_UNIQUE_HASH_LENGTH + pointer); + } + keys+=uniques; /* Each unique has 1 key */ + key_segs+=uniques; /* Each unique has 1 key seg */ + + base_pos=(MARIA_STATE_INFO_SIZE + keys * MARIA_STATE_KEY_SIZE + + key_segs * MARIA_STATE_KEYSEG_SIZE); + info_length= base_pos+(uint) (MARIA_BASE_INFO_SIZE+ + keys * MARIA_KEYDEF_SIZE+ + uniques * MARIA_UNIQUEDEF_SIZE + + (key_segs + unique_key_parts)*HA_KEYSEG_SIZE+ + columns*(MARIA_COLUMNDEF_SIZE + 2)); + + if (encrypted) + { + share.base.extra_options|= MA_EXTRA_OPTIONS_ENCRYPTED; + + /* store crypt data in info */ + info_length+= ma_crypt_get_file_length(); + } + + if (insert_order) + { + share.base.extra_options|= MA_EXTRA_OPTIONS_INSERT_ORDER; + } + + share.state.state.key_file_length= MY_ALIGN(info_length, maria_block_size); + DBUG_PRINT("info", ("info_length: %u", info_length)); + /* There are only 16 bits for the total header length. */ + if (share.state.state.key_file_length > 65535) + { + my_printf_error(HA_WRONG_CREATE_OPTION, + "Aria table '%s' has too many columns and/or " + "indexes and/or unique constraints.", + MYF(0), name + dirname_length(name)); + my_errno= HA_WRONG_CREATE_OPTION; + goto err_no_lock; + } + + bmove(share.state.header.file_version, maria_file_magic, 4); + ci->old_options=options | (ci->old_options & HA_OPTION_TEMP_COMPRESS_RECORD ? + HA_OPTION_COMPRESS_RECORD | + HA_OPTION_TEMP_COMPRESS_RECORD: 0); + mi_int2store(share.state.header.options,ci->old_options); + mi_int2store(share.state.header.header_length,info_length); + mi_int2store(share.state.header.state_info_length,MARIA_STATE_INFO_SIZE); + mi_int2store(share.state.header.base_info_length,MARIA_BASE_INFO_SIZE); + mi_int2store(share.state.header.base_pos,base_pos); + share.state.header.data_file_type= share.data_file_type= datafile_type; + share.state.header.org_data_file_type= org_datafile_type; + share.state.header.not_used= 0; + + share.state.dellink = HA_OFFSET_ERROR; + share.state.first_bitmap_with_space= 0; +#ifdef MARIA_EXTERNAL_LOCKING + share.state.process= (ulong) getpid(); +#endif + share.state.version= (ulong) time((time_t*) 0); + share.state.sortkey= (ushort) ~0; + share.state.auto_increment=ci->auto_increment; + share.options=options; + share.base.rec_reflength=pointer; + share.base.block_size= maria_block_size; + share.base.language= (ci->language ? ci->language : + default_charset_info->number); + + /* + Get estimate for index file length (this may be wrong for FT keys) + This is used for pointers to other key pages. + */ + tmp= (tot_length / maria_block_size + keys * MARIA_INDEX_BLOCK_MARGIN); + + /* + use maximum of key_file_length we calculated and key_file_length value we + got from MAI file header (see also mariapack.c:save_state) + */ + share.base.key_reflength= + maria_get_pointer_length(MY_MAX(ci->key_file_length,tmp),3); + share.base.keys= share.state.header.keys= keys; + share.state.header.uniques= uniques; + share.state.header.fulltext_keys= fulltext_keys; + mi_int2store(share.state.header.key_parts,key_segs); + mi_int2store(share.state.header.unique_key_parts,unique_key_parts); + + maria_set_all_keys_active(share.state.key_map, keys); + + share.base.keystart = share.state.state.key_file_length; + share.base.max_key_block_length= maria_block_size; + share.base.max_key_length=ALIGN_SIZE(max_key_length+4); + share.base.records=ci->max_rows; + share.base.reloc= ci->reloc_rows; + share.base.reclength=real_reclength; + share.base.pack_reclength= reclength + MY_TEST(options & HA_OPTION_CHECKSUM); + share.base.max_pack_length=pack_reclength; + share.base.min_pack_length=min_pack_length; + share.base.pack_bytes= pack_bytes; + share.base.fields= columns; + share.base.pack_fields= packed; + + if (share.data_file_type == BLOCK_RECORD) + { + /* + we are going to create a first bitmap page, set data_file_length + to reflect this, before the state goes to disk + */ + share.state.state.data_file_length= maria_block_size; + /* Add length of packed fields + length */ + share.base.pack_reclength+= share.base.max_field_lengths+3; + share.base.max_pack_length= share.base.pack_reclength; + + /* Adjust max_pack_length, to be used if we have short rows */ + if (share.base.max_pack_length < maria_block_size) + { + share.base.max_pack_length+= FLAG_SIZE; + if (ci->transactional) + share.base.max_pack_length+= TRANSID_SIZE * 2; + } + } + + /* max_data_file_length and max_key_file_length are recalculated on open */ + if (tmp_table) + share.base.max_data_file_length= (my_off_t) ci->data_file_length; + else if (ci->transactional && translog_status == TRANSLOG_OK && + !maria_in_recovery) + { + /* + we have checked translog_inited above, because maria_chk may call us + (via maria_recreate_table()) and it does not have a log. + */ + sync_dir= MY_SYNC_DIR; + /* + If crash between _ma_state_info_write_sub() and + _ma_update_state__lsns_sub(), table should be ignored by Recovery (or + old REDOs would fail), so we cannot let LSNs be 0: + */ + share.state.skip_redo_lsn= share.state.is_of_horizon= + share.state.create_rename_lsn= LSN_MAX; + /* + We have to mark the table as not movable as the table will contain the + maria_uuid and create_rename_lsn + */ + share.state.changed|= STATE_NOT_MOVABLE; + } + + if (datafile_type == DYNAMIC_RECORD) + { + share.base.min_block_length= + (share.base.pack_reclength+3 < MARIA_EXTEND_BLOCK_LENGTH && + ! share.base.blobs) ? + MY_MAX(share.base.pack_reclength,MARIA_MIN_BLOCK_LENGTH) : + MARIA_EXTEND_BLOCK_LENGTH; + } + else if (datafile_type == STATIC_RECORD) + share.base.min_block_length= share.base.pack_reclength; + + if (! (flags & HA_DONT_TOUCH_DATA)) + share.state.create_time= time((time_t*) 0); + + if (!internal_table) + mysql_mutex_lock(&THR_LOCK_maria); + + /* + NOTE: For test_if_reopen() we need a real path name. Hence we need + MY_RETURN_REAL_PATH for every fn_format(filename, ...). + */ + if (ci->index_file_name) + { + char *iext= strrchr(ci->index_file_name, '.'); + int have_iext= iext && !strcmp(iext, MARIA_NAME_IEXT); + if (tmp_table) + { + char *path; + /* chop off the table name, tempory tables use generated name */ + if ((path= strrchr(ci->index_file_name, FN_LIBCHAR))) + *path= '\0'; + fn_format(kfilename, name, ci->index_file_name, MARIA_NAME_IEXT, + MY_REPLACE_DIR | MY_UNPACK_FILENAME | + MY_RETURN_REAL_PATH | MY_APPEND_EXT); + } + else + { + fn_format(kfilename, ci->index_file_name, "", MARIA_NAME_IEXT, + MY_UNPACK_FILENAME | MY_RETURN_REAL_PATH | + (have_iext ? MY_REPLACE_EXT : MY_APPEND_EXT)); + } + fn_format(klinkname, name, "", MARIA_NAME_IEXT, + MY_UNPACK_FILENAME|MY_APPEND_EXT); + klinkname_ptr= klinkname; + /* + Don't create the table if the link or file exists to ensure that one + doesn't accidently destroy another table. + Don't sync dir now if the data file has the same path. + */ + create_flag= + (ci->data_file_name && + !strcmp(ci->index_file_name, ci->data_file_name)) ? 0 : sync_dir; + } + else + { + char *iext= strrchr(name, '.'); + int have_iext= iext && !strcmp(iext, MARIA_NAME_IEXT); + fn_format(kfilename, name, "", MARIA_NAME_IEXT, MY_UNPACK_FILENAME | + (internal_table ? 0 : MY_RETURN_REAL_PATH) | + (have_iext ? MY_REPLACE_EXT : MY_APPEND_EXT)); + klinkname_ptr= NullS; + /* + Replace the current file. + Don't sync dir now if the data file has the same path. + */ + create_flag= (flags & HA_CREATE_KEEP_FILES) ? 0 : MY_DELETE_OLD; + create_flag|= (!ci->data_file_name ? 0 : sync_dir); + } + + /* + If a MRG_MARIA table is in use, the mapped MARIA tables are open, + but no entry is made in the table cache for them. + A TRUNCATE command checks for the table in the cache only and could + be fooled to believe, the table is not open. + Pull the emergency brake in this situation. (Bug #8306) + + + NOTE: The filename is compared against unique_file_name of every + open table. Hence we need a real path here. + */ + if (!internal_table && _ma_test_if_reopen(kfilename)) + { + my_printf_error(HA_ERR_TABLE_EXIST, "Aria table '%s' is in use " + "(most likely by a MERGE table). Try FLUSH TABLES.", + MYF(0), name + dirname_length(name)); + my_errno= HA_ERR_TABLE_EXIST; + goto err; + } + + if ((file= mysql_file_create_with_symlink(key_file_kfile, klinkname_ptr, + kfilename, 0, create_mode, + MYF(common_flag|create_flag))) < 0) + goto err; + errpos=1; + + DBUG_PRINT("info", ("write state info and base info")); + if (_ma_state_info_write_sub(file, &share.state, + MA_STATE_INFO_WRITE_FULL_INFO) || + _ma_base_info_write(file, &share.base)) + goto err; + DBUG_PRINT("info", ("base_pos: %d base_info_size: %d", + base_pos, MARIA_BASE_INFO_SIZE)); + DBUG_ASSERT(mysql_file_tell(file,MYF(0)) == base_pos+ MARIA_BASE_INFO_SIZE); + + /* Write key and keyseg definitions */ + DBUG_PRINT("info", ("write key and keyseg definitions")); + for (i=0 ; i < share.base.keys - uniques; i++) + { + uint sp_segs=(keydefs[i].flag & HA_SPATIAL) ? 2*SPDIMS : 0; + + if (_ma_keydef_write(file, &keydefs[i])) + goto err; + for (j=0 ; j < keydefs[i].keysegs-sp_segs ; j++) + if (_ma_keyseg_write(file, &keydefs[i].seg[j])) + goto err; +#ifdef HAVE_SPATIAL + for (j=0 ; j < sp_segs ; j++) + { + HA_KEYSEG sseg; + sseg.type=SPTYPE; + sseg.language= 7; /* Binary */ + sseg.null_bit=0; + sseg.bit_start=0; + sseg.bit_length= 0; + sseg.bit_pos= 0; + sseg.length=SPLEN; + sseg.null_pos=0; + sseg.start=j*SPLEN; + sseg.flag= HA_SWAP_KEY; + if (_ma_keyseg_write(file, &sseg)) + goto err; + } +#endif + } + /* Create extra keys for unique definitions */ + offset= real_reclength - uniques*MARIA_UNIQUE_HASH_LENGTH; + bzero((char*) &tmp_keydef,sizeof(tmp_keydef)); + bzero((char*) &tmp_keyseg,sizeof(tmp_keyseg)); + for (i=0; i < uniques ; i++) + { + tmp_keydef.keysegs=1; + tmp_keydef.block_length= (uint16) maria_block_size; + tmp_keydef.keylength= MARIA_UNIQUE_HASH_LENGTH + pointer; + tmp_keydef.minlength=tmp_keydef.maxlength=tmp_keydef.keylength; + tmp_keyseg.type= MARIA_UNIQUE_HASH_TYPE; + tmp_keyseg.length= MARIA_UNIQUE_HASH_LENGTH; + tmp_keyseg.start= offset; + offset+= MARIA_UNIQUE_HASH_LENGTH; + if (_ma_keydef_write(file,&tmp_keydef) || + _ma_keyseg_write(file,(&tmp_keyseg))) + goto err; + } + + /* Save unique definition */ + DBUG_PRINT("info", ("write unique definitions")); + for (i=0 ; i < share.state.header.uniques ; i++) + { + HA_KEYSEG *keyseg_end; + keyseg= uniquedefs[i].seg; + if (_ma_uniquedef_write(file, &uniquedefs[i])) + goto err; + for (keyseg= uniquedefs[i].seg, keyseg_end= keyseg+ uniquedefs[i].keysegs; + keyseg < keyseg_end; + keyseg++) + { + switch (keyseg->type) { + case HA_KEYTYPE_VARTEXT1: + case HA_KEYTYPE_VARTEXT2: + case HA_KEYTYPE_VARBINARY1: + case HA_KEYTYPE_VARBINARY2: + if (!(keyseg->flag & HA_BLOB_PART)) + { + keyseg->flag|= HA_VAR_LENGTH_PART; + keyseg->bit_start= ((keyseg->type == HA_KEYTYPE_VARTEXT1 || + keyseg->type == HA_KEYTYPE_VARBINARY1) ? + 1 : 2); + } + break; + default: + DBUG_ASSERT((keyseg->flag & HA_VAR_LENGTH_PART) == 0); + break; + } + if (_ma_keyseg_write(file, keyseg)) + goto err; + } + } + DBUG_PRINT("info", ("write field definitions")); + if (datafile_type == BLOCK_RECORD) + { + /* Store columns in a more efficent order */ + MARIA_COLUMNDEF **col_order, **pos; + if (!(col_order= (MARIA_COLUMNDEF**) my_malloc(PSI_INSTRUMENT_ME, + share.base.fields * + sizeof(MARIA_COLUMNDEF*), + common_flag))) + goto err; + for (column= columndef, pos= col_order ; + column != end_column ; + column++, pos++) + *pos= column; + qsort(col_order, share.base.fields, sizeof(*col_order), + (qsort_cmp) compare_columns); + for (i=0 ; i < share.base.fields ; i++) + { + column_array[col_order[i]->column_nr]= i; + if (_ma_columndef_write(file, col_order[i])) + { + my_free(col_order); + goto err; + } + } + my_free(col_order); + } + else + { + for (i=0 ; i < share.base.fields ; i++) + { + column_array[i]= (uint16) i; + if (_ma_columndef_write(file, &columndef[i])) + goto err; + } + } + if (_ma_column_nr_write(file, column_array, columns)) + goto err; + + if (encrypted) + { + DBUG_ASSERT(share.data_file_name.length == 0); + share.data_file_name.str= (char*) name; /* For error reporting */ + if (ma_crypt_create(&share) || + ma_crypt_write(&share, file)) + goto err; + } + + if ((kfile_size_before_extension= mysql_file_tell(file,MYF(0))) == MY_FILEPOS_ERROR) + goto err; +#ifndef DBUG_OFF + if (kfile_size_before_extension != info_length) + DBUG_PRINT("warning",("info_length: %u != used_length: %u", + info_length, (uint)kfile_size_before_extension)); +#endif + + if (sync_dir) + { + /* + we log the first bytes and then the size to which we extend; this is + a log of about 1 KB of mostly zeroes if this is a small table. + */ + char empty_string[]= ""; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 4]; + translog_size_t total_rec_length= 0; + uint k; + LSN lsn; + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= 1 + 2 + 2 + + (uint) kfile_size_before_extension; + /* we are needing maybe 64 kB, so don't use the stack */ + log_data= my_malloc(PSI_INSTRUMENT_ME, + log_array[TRANSLOG_INTERNAL_PARTS + 1].length, MYF(0)); + if ((log_data == NULL) || + mysql_file_pread(file, 1 + 2 + 2 + log_data, + (size_t) kfile_size_before_extension, 0, MYF(MY_NABP))) + goto err; + /* + remember if the data file was created or not, to know if Recovery can + do it or not, in the future + */ + log_data[0]= MY_TEST(flags & HA_DONT_TOUCH_DATA); + int2store(log_data + 1, kfile_size_before_extension); + int2store(log_data + 1 + 2, share.base.keystart); + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= (uchar *)name; + /* we store the end-zero, for Recovery to just pass it to my_create() */ + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= strlen(name) + 1; + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= log_data; + /* symlink description is also needed for re-creation by Recovery: */ + { + const char *s= ci->data_file_name ? ci->data_file_name : empty_string; + log_array[TRANSLOG_INTERNAL_PARTS + 2].str= (uchar*)s; + log_array[TRANSLOG_INTERNAL_PARTS + 2].length= strlen(s) + 1; + s= ci->index_file_name ? ci->index_file_name : empty_string; + log_array[TRANSLOG_INTERNAL_PARTS + 3].str= (uchar*)s; + log_array[TRANSLOG_INTERNAL_PARTS + 3].length= strlen(s) + 1; + } + for (k= TRANSLOG_INTERNAL_PARTS; + k < (sizeof(log_array)/sizeof(log_array[0])); k++) + total_rec_length+= (translog_size_t) log_array[k].length; + /** + For this record to be of any use for Recovery, we need the upper + MySQL layer to be crash-safe, which it is not now (that would require + work using the ddl_log of sql/sql_table.cc); when it is, we should + reconsider the moment of writing this log record (before or after op, + under THR_LOCK_maria or not...), how to use it in Recovery. + For now this record can serve when we apply logs to a backup, + so we sync it. This happens before the data file is created. If the + data file was created before, and we crashed before writing the log + record, at restart the table may be used, so we would not have a + trustable history in the log (impossible to apply this log to a + backup). The way we do it, if we crash before writing the log record + then there is no data file and the table cannot be used. + @todo Note that in case of TRUNCATE TABLE we also come here; for + Recovery to be able to finish TRUNCATE TABLE, instead of leaving a + half-truncated table, we should log the record at start of + maria_create(); for that we shouldn't write to the index file but to a + buffer (DYNAMIC_STRING), put the buffer into the record, then put the + buffer into the index file (so, change _ma_keydef_write() etc). That + would also enable Recovery to finish a CREATE TABLE. The final result + would be that we would be able to finish what the SQL layer has asked + for: it would be atomic. + When in CREATE/TRUNCATE (or DROP or RENAME or REPAIR) we have not + called external_lock(), so have no TRN. It does not matter, as all + these operations are non-transactional and sync their files. + */ + trnman_init_tmp_trn_for_logging_trid(&tmp_transaction_object); + if (unlikely(translog_write_record(&lsn, + LOGREC_REDO_CREATE_TABLE, + &tmp_transaction_object, NULL, + total_rec_length, + sizeof(log_array)/sizeof(log_array[0]), + log_array, NULL, NULL) || + translog_flush(lsn))) + goto err; + share.kfile.file= file; + DBUG_EXECUTE_IF("maria_flush_whole_log", + { + DBUG_PRINT("maria_flush_whole_log", ("now")); + translog_flush(translog_get_horizon()); + }); + DBUG_EXECUTE_IF("maria_crash_create_table", + { + DBUG_PRINT("maria_crash_create_table", ("now")); + DBUG_SUICIDE(); + }); + /* + store LSN into file, needed for Recovery to not be confused if a + DROP+CREATE happened (applying REDOs to the wrong table). + */ + if (_ma_update_state_lsns_sub(&share, lsn, tmp_transaction_object.trid, + FALSE, TRUE)) + goto err; + my_free(log_data); + log_data= 0; + } + + if (!(flags & HA_DONT_TOUCH_DATA)) + { + if (ci->data_file_name) + { + char *dext= strrchr(ci->data_file_name, '.'); + int have_dext= dext && !strcmp(dext, MARIA_NAME_DEXT); + + if (tmp_table) + { + char *path; + /* chop off the table name, tempory tables use generated name */ + if ((path= strrchr(ci->data_file_name, FN_LIBCHAR))) + *path= '\0'; + fn_format(dfilename, name, ci->data_file_name, MARIA_NAME_DEXT, + MY_REPLACE_DIR | MY_UNPACK_FILENAME | MY_APPEND_EXT); + } + else + { + fn_format(dfilename, ci->data_file_name, "", MARIA_NAME_DEXT, + MY_UNPACK_FILENAME | + (have_dext ? MY_REPLACE_EXT : MY_APPEND_EXT)); + } + fn_format(dlinkname, name, "",MARIA_NAME_DEXT, + MY_UNPACK_FILENAME | MY_APPEND_EXT); + dlinkname_ptr= dlinkname; + create_flag=0; + } + else + { + fn_format(dfilename,name,"", MARIA_NAME_DEXT, + MY_UNPACK_FILENAME | MY_APPEND_EXT); + create_flag= (flags & HA_CREATE_KEEP_FILES) ? 0 : MY_DELETE_OLD; + } + ma_debug_crash_here("storage_engine_middle_of_create"); + if ((dfile= + mysql_file_create_with_symlink(key_file_dfile, dlinkname_ptr, + dfilename, 0, create_mode, + MYF(common_flag | create_flag | sync_dir))) < 0) + goto err; + errpos=3; + + if (_ma_initialize_data_file(&share, dfile)) + goto err; + } + + /* Enlarge files */ + DBUG_PRINT("info", ("enlarge to keystart: %lu", + (ulong) share.base.keystart)); + if (mysql_file_chsize(file,(ulong) share.base.keystart,0,MYF(0))) + goto err; + + if (!internal_table && sync_dir && mysql_file_sync(file, MYF(0))) + goto err; + + if (! (flags & HA_DONT_TOUCH_DATA)) + { +#ifdef USE_RELOC + if (mysql_file_chsize(key_file_dfile, dfile, + share.base.min_pack_length*ci->reloc_rows,0,MYF(0))) + goto err; +#endif + if (!internal_table && sync_dir && mysql_file_sync(dfile, MYF(0))) + goto err; + if (mysql_file_close(dfile,MYF(0))) + goto err; + } + if (!internal_table) + mysql_mutex_unlock(&THR_LOCK_maria); + res= 0; + my_free((char*) rec_per_key_part); + ma_crypt_free(&share); + errpos=0; + if (mysql_file_close(file,MYF(0))) + res= my_errno; + DBUG_RETURN(res); + +err: + if (!internal_table) + mysql_mutex_unlock(&THR_LOCK_maria); + +err_no_lock: + save_errno=my_errno; + switch (errpos) { + case 3: + mysql_file_close(dfile, MYF(0)); + if (! (flags & HA_DONT_TOUCH_DATA)) + { + mysql_file_delete(key_file_dfile, dfilename, MYF(sync_dir)); + if (dlinkname_ptr) + mysql_file_delete(key_file_dfile, dlinkname_ptr, MYF(sync_dir)); + } + /* fall through */ + case 1: + mysql_file_close(file, MYF(0)); + if (! (flags & HA_DONT_TOUCH_DATA)) + { + mysql_file_delete(key_file_kfile, kfilename, MYF(sync_dir)); + if (klinkname_ptr) + mysql_file_delete(key_file_kfile, klinkname_ptr, MYF(sync_dir)); + } + } + ma_crypt_free(&share); + my_free(log_data); + my_free(rec_per_key_part); + DBUG_RETURN(my_errno=save_errno); /* return the fatal errno */ +} + + +uint maria_get_pointer_length(ulonglong file_length, uint def) +{ + DBUG_ASSERT(def >= 2 && def <= 7); + if (file_length) /* If not default */ + { +#ifdef NOT_YET_READY_FOR_8_BYTE_POINTERS + if (file_length >= (1ULL << 56)) + def=8; + else +#endif + if (file_length >= (1ULL << 48)) + def=7; + else if (file_length >= (1ULL << 40)) + def=6; + else if (file_length >= (1ULL << 32)) + def=5; + else if (file_length >= (1ULL << 24)) + def=4; + else if (file_length >= (1ULL << 16)) + def=3; + else + def=2; + } + return def; +} + + +/* + Sort columns for records-in-block + + IMPLEMENTATION + Sort columns in following order: + + Fixed size, not null columns + Fixed length, null fields + Numbers (zero fill fields) + Variable length fields (CHAR, VARCHAR) according to length + Blobs + + For same kind of fields, keep fields in original order +*/ + +static inline int sign(long a) +{ + return a < 0 ? -1 : (a > 0 ? 1 : 0); +} + + +static int compare_columns(MARIA_COLUMNDEF **a_ptr, MARIA_COLUMNDEF **b_ptr) +{ + MARIA_COLUMNDEF *a= *a_ptr, *b= *b_ptr; + enum en_fieldtype a_type, b_type; + + a_type= (a->type == FIELD_CHECK) ? FIELD_NORMAL : a->type; + b_type= (b->type == FIELD_CHECK) ? FIELD_NORMAL : b->type; + + if (a_type == FIELD_NORMAL && !a->null_bit) + { + if (b_type != FIELD_NORMAL || b->null_bit) + return -1; + return sign((long) a->offset - (long) b->offset); + } + if (b_type == FIELD_NORMAL && !b->null_bit) + return 1; + if (a_type == b_type) + return sign((long) a->offset - (long) b->offset); + if (a_type == FIELD_NORMAL) + return -1; + if (b_type == FIELD_NORMAL) + return 1; + if (a_type == FIELD_SKIP_ZERO) + return -1; + if (b_type == FIELD_SKIP_ZERO) + return 1; + if (a->type != FIELD_BLOB && b->type != FIELD_BLOB) + if (a->length != b->length) + return sign((long) a->length - (long) b->length); + if (a_type == FIELD_BLOB) + return 1; + if (b_type == FIELD_BLOB) + return -1; + return sign((long) a->offset - (long) b->offset); +} + + +/** + @brief Initialize data file + + @note + In BLOCK_RECORD, a freshly created datafile is one page long; while in + other formats it is 0-byte long. + */ + +int _ma_initialize_data_file(MARIA_SHARE *share, File dfile) +{ + if (share->data_file_type == BLOCK_RECORD) + { + share->bitmap.block_size= share->base.block_size; + share->bitmap.file.file = dfile; + return _ma_bitmap_create_first(share); + } + return 0; +} + + +/** + @brief Writes create_rename_lsn, skip_redo_lsn and is_of_horizon to disk, + can force. + + This is for special cases where: + - we don't want to write the full state to disk (so, not call + _ma_state_info_write()) because some parts of the state may be + currently inconsistent, or because it would be overkill + - we must sync these LSNs immediately for correctness. + It acquires intern_lock to protect the LSNs and state write. + + @param share table's share + @param lsn LSN to write to log files + @param create_trid Trid to be used as state.create_trid + @param do_sync if the write should be forced to disk + @param update_create_rename_lsn if this LSN should be updated or not + + @return Operation status + @retval 0 ok + @retval 1 error (disk problem) +*/ + +int _ma_update_state_lsns(MARIA_SHARE *share, LSN lsn, TrID create_trid, + my_bool do_sync, my_bool update_create_rename_lsn) +{ + int res; + DBUG_ENTER("_ma_update_state_lsns"); + mysql_mutex_lock(&share->intern_lock); + res= _ma_update_state_lsns_sub(share, lsn, create_trid, do_sync, + update_create_rename_lsn); + mysql_mutex_unlock(&share->intern_lock); + DBUG_RETURN(res); +} + + +/** + @brief Writes create_rename_lsn, skip_redo_lsn and is_of_horizon to disk, + can force. + + Shortcut of _ma_update_state_lsns() when we know that intern_lock is not + needed (when creating a table or opening it for the first time). + + @param share table's share + @param lsn LSN to write to state; if LSN_IMPOSSIBLE, write + a LOGREC_IMPORTED_TABLE and use its LSN as lsn. + @param create_trid Trid to be used as state.create_trid + @param do_sync if the write should be forced to disk + @param update_create_rename_lsn if this LSN should be updated or not + + @return Operation status + @retval 0 ok + @retval 1 error (disk problem) +*/ + +#if defined(_MSC_VER) && (_MSC_VER == 1310) +/* + Visual Studio 2003 compiler produces internal compiler error + in this function. Disable optimizations to workaround. +*/ +#pragma optimize("",off) +#endif +int _ma_update_state_lsns_sub(MARIA_SHARE *share, LSN lsn, TrID create_trid, + my_bool do_sync, + my_bool update_create_rename_lsn) +{ + uchar buf[LSN_STORE_SIZE * 3], *ptr; + uchar trid_buff[8]; + File file= share->kfile.file; + DBUG_ENTER("_ma_update_state_lsns_sub"); + DBUG_ASSERT(file >= 0); + CRASH_IF_S3_TABLE(share); + + if (lsn == LSN_IMPOSSIBLE) + { + int res; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 1]; + /* table name is logged only for information */ + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= + (uchar *)(share->open_file_name.str); + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= + share->open_file_name.length + 1; + if ((res= translog_write_record(&lsn, LOGREC_IMPORTED_TABLE, + &dummy_transaction_object, NULL, + (translog_size_t) + log_array[TRANSLOG_INTERNAL_PARTS + + 0].length, + sizeof(log_array)/sizeof(log_array[0]), + log_array, NULL, NULL))) + DBUG_RETURN(res); + } + + for (ptr= buf; ptr < (buf + sizeof(buf)); ptr+= LSN_STORE_SIZE) + lsn_store(ptr, lsn); + share->state.skip_redo_lsn= share->state.is_of_horizon= lsn; + share->state.create_trid= create_trid; + mi_int8store(trid_buff, create_trid); + + /* + Update create_rename_lsn if update was requested or if the old one had an + impossible value. + */ + if (update_create_rename_lsn || + (share->state.create_rename_lsn > lsn && lsn != LSN_IMPOSSIBLE)) + { + share->state.create_rename_lsn= lsn; + if (share->id != 0) + { + /* + If OP is the operation which is calling us, if table is later written, + we could see in the log: + FILE_ID ... REDO_OP ... REDO_INSERT. + (that can happen in real life at least with OP=REPAIR). + As FILE_ID will be ignored by Recovery because it is < + create_rename_lsn, REDO_INSERT would be ignored too, wrongly. + To avoid that, we force a LOGREC_FILE_ID to be logged at next write: + */ + translog_deassign_id_from_share(share); + } + } + else + lsn_store(buf, share->state.create_rename_lsn); + DBUG_RETURN(my_pwrite(file, buf, sizeof(buf), + sizeof(share->state.header) + + MARIA_FILE_CREATE_RENAME_LSN_OFFSET, MYF(MY_NABP)) || + my_pwrite(file, trid_buff, sizeof(trid_buff), + sizeof(share->state.header) + + MARIA_FILE_CREATE_TRID_OFFSET, MYF(MY_NABP)) || + (do_sync && mysql_file_sync(file, MYF(0)))); +} +#if defined(_MSC_VER) && (_MSC_VER == 1310) +#pragma optimize("",on) +#endif /*VS2003 compiler bug workaround*/ |