diff options
Diffstat (limited to '')
| -rw-r--r-- | storage/maria/ma_write.c | 2503 |
1 files changed, 2503 insertions, 0 deletions
diff --git a/storage/maria/ma_write.c b/storage/maria/ma_write.c new file mode 100644 index 00000000..95cc1203 --- /dev/null +++ b/storage/maria/ma_write.c @@ -0,0 +1,2503 @@ +/* Copyright (C) 2004-2008 MySQL AB & MySQL Finland AB & TCX DataKonsult AB + Copyright (C) 2008-2009 Sun Microsystems, Inc. + + 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 */ + +/* Write a row to a MARIA table */ + +#include "ma_fulltext.h" +#include "ma_rt_index.h" +#include "trnman.h" +#include "ma_key_recover.h" +#include "ma_blockrec.h" + + /* Functions declared in this file */ + +static int w_search(MARIA_HA *info, uint32 comp_flag, + MARIA_KEY *key, my_off_t page, + MARIA_PAGE *father_page, uchar *father_keypos, + my_bool insert_last); +static int _ma_balance_page(MARIA_HA *info, MARIA_KEYDEF *keyinfo, + MARIA_KEY *key, MARIA_PAGE *curr_page, + MARIA_PAGE *father_page, + uchar *father_key_pos, MARIA_KEY_PARAM *s_temp); +static uchar *_ma_find_last_pos(MARIA_KEY *int_key, + MARIA_PAGE *page, uchar **after_key); +static my_bool _ma_ck_write_tree(register MARIA_HA *info, MARIA_KEY *key); +static my_bool _ma_ck_write_btree(register MARIA_HA *info, MARIA_KEY *key); +static my_bool _ma_ck_write_btree_with_log(MARIA_HA *, MARIA_KEY *, my_off_t *, + uint32); +static my_bool _ma_log_split(MARIA_PAGE *page, uint org_length, + uint new_length, + const uchar *key_pos, + uint key_length, int move_length, + enum en_key_op prefix_or_suffix, + const uchar *data, uint data_length, + uint changed_length); +static my_bool _ma_log_del_prefix(MARIA_PAGE *page, + uint org_length, uint new_length, + const uchar *key_pos, uint key_length, + int move_length); +static my_bool _ma_log_key_middle(MARIA_PAGE *page, + uint new_length, + uint data_added_first, + uint data_changed_first, + uint data_deleted_last, + const uchar *key_pos, + uint key_length, int move_length); + +/* + @brief Default handler for returing position to new row + + @note + This is only called for non transactional tables and not for block format + which is why we use info->state here. +*/ + +MARIA_RECORD_POS _ma_write_init_default(MARIA_HA *info, + const uchar *record + __attribute__((unused))) +{ + return ((info->s->state.dellink != HA_OFFSET_ERROR && + !info->append_insert_at_end) ? + info->s->state.dellink : + info->state->data_file_length); +} + +my_bool _ma_write_abort_default(MARIA_HA *info __attribute__((unused))) +{ + return 0; +} + + +/* Write new record to a table */ + +int maria_write(MARIA_HA *info, const uchar *record) +{ + MARIA_SHARE *share= info->s; + uint i; + int save_errno; + MARIA_RECORD_POS filepos, oldpos= info->cur_row.lastpos; + uchar *buff; + my_bool lock_tree= share->lock_key_trees; + my_bool fatal_error; + MARIA_KEYDEF *keyinfo; + DBUG_ENTER("maria_write"); + DBUG_PRINT("enter",("index_file: %d data_file: %d", + share->kfile.file, info->dfile.file)); + + DBUG_EXECUTE_IF("maria_pretend_crashed_table_on_usage", + _ma_print_error(info, HA_ERR_CRASHED, 0); + DBUG_RETURN(my_errno= HA_ERR_CRASHED);); + if (share->options & HA_OPTION_READ_ONLY_DATA) + { + DBUG_RETURN(my_errno=EACCES); + } + if (_ma_readinfo(info,F_WRLCK,1)) + DBUG_RETURN(my_errno); + + if ((share->state.changed & STATE_DATA_FILE_FULL) || + (share->base.reloc == (ha_rows) 1 && + share->base.records == (ha_rows) 1 && + share->state.state.records == (ha_rows) 1)) + { /* System file */ + my_errno=HA_ERR_RECORD_FILE_FULL; + goto err2; + } + if (share->state.state.key_file_length >= share->base.margin_key_file_length) + { + my_errno=HA_ERR_INDEX_FILE_FULL; + goto err2; + } + if (_ma_mark_file_changed(share)) + goto err2; + + /* Calculate and check all unique constraints */ + + if (share->state.header.uniques) + { + for (i=0 ; i < share->state.header.uniques ; i++) + { + MARIA_UNIQUEDEF *def= share->uniqueinfo + i; + ha_checksum unique_hash= _ma_unique_hash(share->uniqueinfo+i,record); + if (maria_is_key_active(share->state.key_map, def->key)) + { + if (_ma_check_unique(info, def, record, + unique_hash, HA_OFFSET_ERROR)) + goto err2; + } + else + maria_unique_store(record+ share->keyinfo[def->key].seg->start, + unique_hash); + } + } + + /* Ensure we don't try to restore auto_increment if it doesn't change */ + info->last_auto_increment= ~(ulonglong) 0; + + if ((info->opt_flag & OPT_NO_ROWS)) + filepos= HA_OFFSET_ERROR; + else + { + /* + This may either calculate a record or, or write the record and return + the record id + */ + if ((filepos= (*share->write_record_init)(info, record)) == + HA_OFFSET_ERROR) + goto err2; + } + + /* Write all keys to indextree */ + buff= info->lastkey_buff2; + for (i=0, keyinfo= share->keyinfo ; i < share->base.keys ; i++, keyinfo++) + { + MARIA_KEY int_key; + if (maria_is_key_active(share->state.key_map, i)) + { + my_bool local_lock_tree= (lock_tree && + !(info->bulk_insert && + is_tree_inited(&info->bulk_insert[i]))); + if (local_lock_tree) + { + mysql_rwlock_wrlock(&keyinfo->root_lock); + keyinfo->version++; + } + if (keyinfo->flag & HA_FULLTEXT ) + { + if (_ma_ft_add(info,i, buff,record,filepos)) + { + if (local_lock_tree) + mysql_rwlock_unlock(&keyinfo->root_lock); + DBUG_PRINT("error",("Got error: %d on write",my_errno)); + goto err; + } + } + else + { + while (keyinfo->ck_insert(info, + (*keyinfo->make_key)(info, &int_key, i, + buff, record, filepos, + info->trn->trid))) + { + TRN *blocker; + DBUG_PRINT("error",("Got error: %d on write",my_errno)); + /* + explicit check to filter out temp tables, they aren't + transactional and don't have a proper TRN so the code + below doesn't work for them. + Also, filter out non-thread maria use, and table modified in + the same transaction. + At last, filter out non-dup-unique errors. + */ + if (!local_lock_tree) + goto err; + if (info->dup_key_trid == info->trn->trid || + my_errno != HA_ERR_FOUND_DUPP_KEY) + { + mysql_rwlock_unlock(&keyinfo->root_lock); + goto err; + } + /* Different TrIDs: table must be transactional */ + DBUG_ASSERT(share->base.born_transactional); + /* + If transactions are disabled, and dup_key_trid is different from + our TrID, it must be ALTER TABLE with dup_key_trid==0 (no + transaction). ALTER TABLE does have MARIA_HA::TRN not dummy but + puts TrID=0 in rows/keys. + */ + DBUG_ASSERT(share->now_transactional || + (info->dup_key_trid == 0)); + blocker= trnman_trid_to_trn(info->trn, info->dup_key_trid); + /* + if blocker TRN was not found, it means that the conflicting + transaction was committed long time ago. It could not be + aborted, as it would have to wait on the key tree lock + to remove the conflicting key it has inserted. + */ + if (!blocker || blocker->commit_trid != ~(TrID)0) + { /* committed */ + if (blocker) + mysql_mutex_unlock(& blocker->state_lock); + mysql_rwlock_unlock(&keyinfo->root_lock); + goto err; + } + mysql_rwlock_unlock(&keyinfo->root_lock); + { + /* running. now we wait */ + WT_RESOURCE_ID rc; + int res; + PSI_stage_info old_stage_info; + + rc.type= &ma_rc_dup_unique; + /* TODO savepoint id when we'll have them */ + rc.value= (intptr)blocker; + res= wt_thd_will_wait_for(info->trn->wt, blocker->wt, & rc); + if (res != WT_OK) + { + mysql_mutex_unlock(& blocker->state_lock); + my_errno= HA_ERR_LOCK_DEADLOCK; + goto err; + } + proc_info_hook(0, &stage_waiting_for_a_resource, &old_stage_info, + __func__, __FILE__, __LINE__); + res= wt_thd_cond_timedwait(info->trn->wt, & blocker->state_lock); + proc_info_hook(0, &old_stage_info, 0, __func__, __FILE__, __LINE__); + + mysql_mutex_unlock(& blocker->state_lock); + if (res != WT_OK) + { + my_errno= res == WT_TIMEOUT ? HA_ERR_LOCK_WAIT_TIMEOUT + : HA_ERR_LOCK_DEADLOCK; + goto err; + } + } + mysql_rwlock_wrlock(&keyinfo->root_lock); +#ifndef MARIA_CANNOT_ROLLBACK + keyinfo->version++; +#endif + } + } + + /* The above changed info->lastkey2. Inform maria_rnext_same(). */ + info->update&= ~HA_STATE_RNEXT_SAME; + + if (local_lock_tree) + mysql_rwlock_unlock(&keyinfo->root_lock); + } + } + if (share->calc_write_checksum) + info->cur_row.checksum= (*share->calc_write_checksum)(info,record); + if (filepos != HA_OFFSET_ERROR) + { + if ((*share->write_record)(info,record)) + goto err; + info->state->checksum+= info->cur_row.checksum; + } + if (!share->now_transactional) + { + if (share->base.auto_key != 0) + { + const HA_KEYSEG *keyseg= share->keyinfo[share->base.auto_key-1].seg; + const uchar *key= record + keyseg->start; + set_if_bigger(share->state.auto_increment, + ma_retrieve_auto_increment(key, keyseg->type)); + } + } + info->state->records++; + info->update= (HA_STATE_CHANGED | HA_STATE_AKTIV | HA_STATE_WRITTEN | + HA_STATE_ROW_CHANGED); + info->row_changes++; + share->state.changed|= STATE_NOT_MOVABLE | STATE_NOT_ZEROFILLED; + info->state->changed= 1; + + info->cur_row.lastpos= oldpos; + _ma_writeinfo(info, WRITEINFO_UPDATE_KEYFILE); + if (info->invalidator != 0) + { + DBUG_PRINT("info", ("invalidator... '%s' (update)", + share->open_file_name.str)); + (*info->invalidator)(share->open_file_name.str); + info->invalidator=0; + } + + /* + Update status of the table. We need to do so after each row write + for the log tables, as we want the new row to become visible to + other threads as soon as possible. We don't lock mutex here + (as it is required by pthread memory visibility rules) as (1) it's + not critical to use outdated share->is_log_table value (2) locking + mutex here for every write is too expensive. + */ + if (share->is_log_table) + _ma_update_status((void*) info); + + DBUG_RETURN(0); + +err: + save_errno= my_errno; + fatal_error= 0; + if (my_errno == HA_ERR_FOUND_DUPP_KEY || + my_errno == HA_ERR_RECORD_FILE_FULL || + my_errno == HA_ERR_LOCK_DEADLOCK || + my_errno == HA_ERR_LOCK_WAIT_TIMEOUT || + my_errno == HA_ERR_NULL_IN_SPATIAL || + my_errno == HA_ERR_OUT_OF_MEM) + { + info->errkey= i < share->base.keys ? (int) i : -1; + /* + We delete keys in the reverse order of insertion. This is the order that + a rollback would do and is important for CLR_ENDs generated by + _ma_ft|ck_delete() and write_record_abort() to work (with any other + order they would cause wrong jumps in the chain). + */ + while ( i-- > 0) + { + if (maria_is_key_active(share->state.key_map, i)) + { + my_bool local_lock_tree= (lock_tree && + !(info->bulk_insert && + is_tree_inited(&info->bulk_insert[i]))); + keyinfo= share->keyinfo + i; + if (local_lock_tree) + mysql_rwlock_wrlock(&keyinfo->root_lock); + /** + @todo RECOVERY BUG + The key deletes below should generate CLR_ENDs + */ + if (keyinfo->flag & HA_FULLTEXT) + { + if (_ma_ft_del(info,i,buff,record,filepos)) + { + fatal_error= 1; + if (local_lock_tree) + mysql_rwlock_unlock(&keyinfo->root_lock); + break; + } + } + else + { + MARIA_KEY key; + if (keyinfo->ck_delete(info, + (*keyinfo->make_key)(info, &key, i, buff, + record, + filepos, + info->trn->trid))) + { + fatal_error= 1; + if (local_lock_tree) + mysql_rwlock_unlock(&keyinfo->root_lock); + break; + } + } + if (local_lock_tree) + mysql_rwlock_unlock(&keyinfo->root_lock); + } + } + } + else + fatal_error= 1; + + if (filepos != HA_OFFSET_ERROR) + { + if ((*share->write_record_abort)(info)) + fatal_error= 1; + } + + if (info->bulk_insert) + { + uint j; + for (j=0 ; j < share->base.keys ; j++) + maria_flush_bulk_insert(info, j); + } + + if (fatal_error) + _ma_set_fatal_error(info, HA_ERR_CRASHED); + + info->update= (HA_STATE_CHANGED | HA_STATE_WRITTEN | HA_STATE_ROW_CHANGED); + my_errno=save_errno; +err2: + save_errno=my_errno; + DBUG_ASSERT(save_errno); + if (!save_errno) + save_errno= HA_ERR_INTERNAL_ERROR; /* Should never happen */ + DBUG_PRINT("error", ("got error: %d", save_errno)); + _ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE); + DBUG_RETURN(my_errno=save_errno); +} /* maria_write */ + + +/* + Write one key to btree + + TODO + Remove this function and have bulk insert change keyinfo->ck_insert + to point to the right function +*/ + +my_bool _ma_ck_write(MARIA_HA *info, MARIA_KEY *key) +{ + DBUG_ENTER("_ma_ck_write"); + + if (info->bulk_insert && + is_tree_inited(&info->bulk_insert[key->keyinfo->key_nr])) + { + DBUG_RETURN(_ma_ck_write_tree(info, key)); + } + DBUG_RETURN(_ma_ck_write_btree(info, key)); +} /* _ma_ck_write */ + + +/********************************************************************** + Insert key into btree (normal case) +**********************************************************************/ + +static my_bool _ma_ck_write_btree(MARIA_HA *info, MARIA_KEY *key) +{ + my_bool error; + MARIA_KEYDEF *keyinfo= key->keyinfo; + my_off_t *root= &info->s->state.key_root[keyinfo->key_nr]; + DBUG_ENTER("_ma_ck_write_btree"); + + error= _ma_ck_write_btree_with_log(info, key, root, + keyinfo->write_comp_flag | key->flag); + if (info->ft1_to_ft2) + { + if (!error) + error= _ma_ft_convert_to_ft2(info, key); + delete_dynamic(info->ft1_to_ft2); + my_free(info->ft1_to_ft2); + info->ft1_to_ft2=0; + } + DBUG_RETURN(error); +} /* _ma_ck_write_btree */ + + +/** + @brief Write a key to the b-tree + + @retval 1 error + @retval 0 ok +*/ + +static my_bool _ma_ck_write_btree_with_log(MARIA_HA *info, MARIA_KEY *key, + my_off_t *root, uint32 comp_flag) +{ + MARIA_SHARE *share= info->s; + LSN lsn= LSN_IMPOSSIBLE; + int error; + my_off_t new_root= *root; + uchar key_buff[MARIA_MAX_KEY_BUFF]; + MARIA_KEY org_key; /* Set/used when now_transactional=TRUE */ + my_bool transactional= share->now_transactional; + DBUG_ENTER("_ma_ck_write_btree_with_log"); + + LINT_INIT_STRUCT(org_key); + + if (transactional) + { + /* Save original value as the key may change */ + org_key= *key; + memcpy(key_buff, key->data, key->data_length + key->ref_length); + } + + error= _ma_ck_real_write_btree(info, key, &new_root, comp_flag); + if (!error && transactional) + { + /* Log the original value */ + *key= org_key; + key->data= key_buff; + error= _ma_write_undo_key_insert(info, key, root, new_root, &lsn); + } + else + { + *root= new_root; + _ma_fast_unlock_key_del(info); + } + _ma_unpin_all_pages_and_finalize_row(info, lsn); + + DBUG_RETURN(error != 0); +} /* _ma_ck_write_btree_with_log */ + + +/** + @brief Write a key to the b-tree + + @retval 1 error + @retval 0 ok +*/ + +my_bool _ma_ck_real_write_btree(MARIA_HA *info, MARIA_KEY *key, my_off_t *root, + uint32 comp_flag) +{ + int error; + DBUG_ENTER("_ma_ck_real_write_btree"); + + /* key_length parameter is used only if comp_flag is SEARCH_FIND */ + if (*root == HA_OFFSET_ERROR || + (error= w_search(info, comp_flag, key, *root, (MARIA_PAGE *) 0, + (uchar*) 0, 1)) > 0) + error= _ma_enlarge_root(info, key, root); + DBUG_RETURN(error != 0); +} /* _ma_ck_real_write_btree */ + + +/** + @brief Make a new root with key as only pointer + + @retval 1 error + @retval 0 ok +*/ + +my_bool _ma_enlarge_root(MARIA_HA *info, MARIA_KEY *key, my_off_t *root) +{ + uint t_length, nod_flag; + MARIA_KEY_PARAM s_temp; + MARIA_SHARE *share= info->s; + MARIA_PINNED_PAGE tmp_page_link, *page_link= &tmp_page_link; + MARIA_KEYDEF *keyinfo= key->keyinfo; + MARIA_PAGE page; + my_bool res= 0; + DBUG_ENTER("_ma_enlarge_root"); + + page.info= info; + page.keyinfo= keyinfo; + page.buff= info->buff; + page.flag= 0; + + nod_flag= (*root != HA_OFFSET_ERROR) ? share->base.key_reflength : 0; + /* Store pointer to prev page if nod */ + _ma_kpointer(info, page.buff + share->keypage_header, *root); + t_length= (*keyinfo->pack_key)(key, nod_flag, (uchar*) 0, + (uchar*) 0, (uchar*) 0, &s_temp); + page.size= share->keypage_header + t_length + nod_flag; + + bzero(page.buff, share->keypage_header); + _ma_store_keynr(share, page.buff, keyinfo->key_nr); + if (nod_flag) + page.flag|= KEYPAGE_FLAG_ISNOD; + if (key->flag & (SEARCH_USER_KEY_HAS_TRANSID | SEARCH_PAGE_KEY_HAS_TRANSID)) + page.flag|= KEYPAGE_FLAG_HAS_TRANSID; + (*keyinfo->store_key)(keyinfo, page.buff + share->keypage_header + + nod_flag, &s_temp); + + /* Mark that info->buff was used */ + info->keyread_buff_used= info->page_changed= 1; + if ((page.pos= _ma_new(info, PAGECACHE_PRIORITY_HIGH, &page_link)) == + HA_OFFSET_ERROR) + DBUG_RETURN(1); + *root= page.pos; + + page_store_info(share, &page); + + /* + Clear unitialized part of page to avoid valgrind/purify warnings + and to get a clean page that is easier to compress and compare with + pages generated with redo + */ + bzero(page.buff + page.size, share->block_size - page.size); + + if (share->now_transactional && _ma_log_new(&page, 1)) + res= 1; + + if (_ma_write_keypage(&page, page_link->write_lock, + PAGECACHE_PRIORITY_HIGH)) + res= 1; + + DBUG_RETURN(res); +} /* _ma_enlarge_root */ + + +/* + Search after a position for a key and store it there + + TODO: + Change this to use pagecache directly instead of creating a copy + of the page. To do this, we must however change write-key-on-page + algorithm to not overwrite the buffer but instead store any overflow + key in a separate buffer. + + @return + @retval -1 error + @retval 0 ok + @retval > 0 Key should be stored in higher tree +*/ + +static int w_search(register MARIA_HA *info, uint32 comp_flag, MARIA_KEY *key, + my_off_t page_pos, + MARIA_PAGE *father_page, uchar *father_keypos, + my_bool insert_last) +{ + int error,flag; + uchar *temp_buff,*keypos,*keybuff; + my_bool was_last_key, buff_alloced; + my_off_t next_page, dup_key_pos; + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *keyinfo= key->keyinfo; + MARIA_PAGE page; + DBUG_ENTER("w_search"); + DBUG_PRINT("enter", ("page: %lu", (ulong) (page_pos/keyinfo->block_length))); + + alloc_on_stack(*info->stack_end_ptr, temp_buff, buff_alloced, + (keyinfo->block_length + keyinfo->max_store_length*3)); + if (!temp_buff) + DBUG_RETURN(1); + + keybuff= temp_buff + (keyinfo->block_length + keyinfo->max_store_length*2); + + if (_ma_fetch_keypage(&page, info, keyinfo, page_pos, PAGECACHE_LOCK_WRITE, + DFLT_INIT_HITS, temp_buff, 0)) + goto err; + + flag= (*keyinfo->bin_search)(key, &page, comp_flag, &keypos, + keybuff, &was_last_key); + if (flag == 0) + { + MARIA_KEY tmp_key; + /* get position to record with duplicated key */ + + tmp_key.keyinfo= keyinfo; + tmp_key.data= keybuff; + + if ((*keyinfo->get_key)(&tmp_key, page.flag, page.node, &keypos)) + dup_key_pos= _ma_row_pos_from_key(&tmp_key); + else + dup_key_pos= HA_OFFSET_ERROR; + + if (keyinfo->flag & HA_FULLTEXT) + { + uint off; + int subkeys; + + get_key_full_length_rdonly(off, keybuff); + subkeys=ft_sintXkorr(keybuff+off); + comp_flag=SEARCH_SAME; + if (subkeys >= 0) + { + /* normal word, one-level tree structure */ + flag=(*keyinfo->bin_search)(key, &page, comp_flag, + &keypos, keybuff, &was_last_key); + } + else + { + /* popular word. two-level tree. going down */ + my_off_t root= dup_key_pos; + MARIA_KEY subkey; + get_key_full_length_rdonly(off, key->data); + subkey.keyinfo= keyinfo= &share->ft2_keyinfo; + subkey.data= key->data + off; + subkey.data_length= key->data_length - off; + subkey.ref_length= key->ref_length; + subkey.flag= key->flag; + + /* we'll modify key entry 'in vivo' */ + keypos-= keyinfo->keylength + page.node; + error= _ma_ck_real_write_btree(info, &subkey, &root, comp_flag); + _ma_dpointer(share, keypos+HA_FT_WLEN, root); + subkeys--; /* should there be underflow protection ? */ + DBUG_ASSERT(subkeys < 0); + ft_intXstore(keypos, subkeys); + if (!error) + { + page_mark_changed(info, &page); + if (_ma_write_keypage(&page, PAGECACHE_LOCK_LEFT_WRITELOCKED, + DFLT_INIT_HITS)) + goto err; + } + stack_alloc_free(temp_buff, buff_alloced); + DBUG_RETURN(error); + } + } + else /* not HA_FULLTEXT, normal HA_NOSAME key */ + { + /* + TODO + When the index will support true versioning - with multiple + identical values in the UNIQUE index, invisible to each other - + the following should be changed to "continue inserting keys, at the + end (of the row or statement) wait". We need to wait on *all* + unique conflicts at once, not one-at-a-time, because we need to + know all blockers in advance, otherwise we'll have incomplete wait-for + graph. + */ + /* + transaction that has inserted the conflicting key may be in progress. + the caller will wait for it to be committed or aborted. + */ + info->dup_key_trid= _ma_trid_from_key(&tmp_key); + info->dup_key_pos= dup_key_pos; + my_errno= HA_ERR_FOUND_DUPP_KEY; + DBUG_PRINT("warning", + ("Duplicate key. dup_key_trid: %lu pos %lu visible: %d", + (ulong) info->dup_key_trid, + (ulong) info->dup_key_pos, + info->trn ? trnman_can_read_from(info->trn, + info->dup_key_trid) : 2)); + goto err; + } + } + if (flag == MARIA_FOUND_WRONG_KEY) + { + my_errno= HA_ERR_CRASHED; + goto err; + } + if (!was_last_key) + insert_last=0; + next_page= _ma_kpos(page.node, keypos); + if (next_page == HA_OFFSET_ERROR || + (error= w_search(info, comp_flag, key, next_page, + &page, keypos, insert_last)) > 0) + { + error= _ma_insert(info, key, &page, keypos, keybuff, + father_page, father_keypos, insert_last); + if (error < 0) + goto err; + page_mark_changed(info, &page); + if (_ma_write_keypage(&page, PAGECACHE_LOCK_LEFT_WRITELOCKED, + DFLT_INIT_HITS)) + goto err; + } + stack_alloc_free(temp_buff, buff_alloced); + DBUG_RETURN(error); +err: + stack_alloc_free(temp_buff, buff_alloced); + DBUG_PRINT("exit",("Error: %d",my_errno)); + DBUG_RETURN(-1); +} /* w_search */ + + +/* + Insert new key. + + SYNOPSIS + _ma_insert() + info Open table information. + keyinfo Key definition information. + key New key + anc_page Key page (beginning) + key_pos Position in key page where to insert. + key_buff Copy of previous key if keys where packed. + father_page position of parent key page in file. + father_key_pos position in parent key page for balancing. + insert_last If to append at end of page. + + DESCRIPTION + Insert new key at right of key_pos. + Note that caller must save anc_buff + + This function writes log records for all changed pages + (Including anc_buff and father page) + + RETURN + < 0 Error. + 0 OK + 1 If key contains key to upper level (from balance page) + 2 If key contains key to upper level (from split space) +*/ + +int _ma_insert(register MARIA_HA *info, MARIA_KEY *key, + MARIA_PAGE *anc_page, uchar *key_pos, uchar *key_buff, + MARIA_PAGE *father_page, uchar *father_key_pos, + my_bool insert_last) +{ + uint a_length, nod_flag, org_anc_length; + int t_length; + uchar *endpos, *prev_key, *anc_buff; + MARIA_KEY_PARAM s_temp; + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *keyinfo= key->keyinfo; + DBUG_ENTER("_ma_insert"); + DBUG_PRINT("enter",("key_pos:%p", key_pos)); + DBUG_EXECUTE("key", _ma_print_key(DBUG_FILE, key);); + + /* + Note that anc_page->size can be bigger then block_size in case of + delete key that caused increase of page length + */ + org_anc_length= a_length= anc_page->size; + nod_flag= anc_page->node; + + anc_buff= anc_page->buff; + endpos= anc_buff+ a_length; + prev_key= (key_pos == anc_buff + share->keypage_header + nod_flag ? + (uchar*) 0 : key_buff); + t_length= (*keyinfo->pack_key)(key, nod_flag, + (key_pos == endpos ? (uchar*) 0 : key_pos), + prev_key, prev_key, &s_temp); +#ifndef DBUG_OFF + if (prev_key && (keyinfo->flag & (HA_BINARY_PACK_KEY | HA_PACK_KEY))) + { + DBUG_DUMP("prev_key", prev_key, _ma_keylength(keyinfo,prev_key)); + } + if (keyinfo->flag & HA_PACK_KEY) + { + DBUG_PRINT("test",("t_length: %d ref_len: %d", + t_length,s_temp.ref_length)); + DBUG_PRINT("test",("n_ref_len: %d n_length: %d key_pos: %p", + s_temp.n_ref_length, s_temp.n_length, s_temp.key)); + } +#endif + if (t_length > 0) + { + if (t_length >= keyinfo->maxlength*2+MARIA_INDEX_OVERHEAD_SIZE) + { + _ma_set_fatal_error(info, HA_ERR_CRASHED); + DBUG_RETURN(-1); + } + bmove_upp(endpos+t_length, endpos, (uint) (endpos-key_pos)); + } + else + { + if (-t_length >= keyinfo->maxlength*2+MARIA_INDEX_OVERHEAD_SIZE) + { + _ma_set_fatal_error(info, HA_ERR_CRASHED); + DBUG_RETURN(-1); + } + bmove(key_pos,key_pos-t_length,(uint) (endpos-key_pos)+t_length); + } + (*keyinfo->store_key)(keyinfo,key_pos,&s_temp); + a_length+=t_length; + + if (key->flag & (SEARCH_USER_KEY_HAS_TRANSID | SEARCH_PAGE_KEY_HAS_TRANSID)) + _ma_mark_page_with_transid(share, anc_page); + + anc_page->size= a_length; + page_store_size(share, anc_page); + + /* + Check if the new key fits totally into the the page + (anc_buff is big enough to contain a full page + one key) + */ + if (a_length <= share->max_index_block_size) + { + if (share->max_index_block_size - a_length < 32 && + (keyinfo->flag & HA_FULLTEXT) && key_pos == endpos && + share->base.key_reflength <= share->rec_reflength && + share->options & (HA_OPTION_PACK_RECORD | HA_OPTION_COMPRESS_RECORD)) + { + /* + Normal word. One-level tree. Page is almost full. + Let's consider converting. + We'll compare 'key' and the first key at anc_buff + */ + const uchar *a= key->data; + const uchar *b= anc_buff + share->keypage_header + nod_flag; + uint alen, blen, ft2len= share->ft2_keyinfo.keylength; + /* the very first key on the page is always unpacked */ + DBUG_ASSERT((*b & 128) == 0); +#if HA_FT_MAXLEN >= 127 + blen= mi_uint2korr(b); b+=2; + When you enable this code, as part of the MyISAM->Maria merge of +ChangeSet@1.2562, 2008-04-09 07:41:40+02:00, serg@janus.mylan +9 -0 + restore ft2 functionality, fix bugs. + Then this will enable two-level fulltext index, which is not totally + recoverable yet. + So remove this text and inform Guilhem so that he fixes the issue. +#else + blen= *b++; +#endif + get_key_length(alen,a); + DBUG_ASSERT(info->ft1_to_ft2==0); + if (alen == blen && + ha_compare_char_varying(keyinfo->seg->charset, + a, alen, + b, blen, + FALSE/*b_is_prefix*/) == 0) + { + /* Yup. converting */ + info->ft1_to_ft2=(DYNAMIC_ARRAY *) + my_malloc(PSI_INSTRUMENT_ME, sizeof(DYNAMIC_ARRAY), MYF(MY_WME)); + my_init_dynamic_array(PSI_INSTRUMENT_ME, info->ft1_to_ft2, ft2len, 300, + 50, MYF(0)); + + /* + Now, adding all keys from the page to dynarray + if the page is a leaf (if not keys will be deleted later) + */ + if (!nod_flag) + { + /* + Let's leave the first key on the page, though, because + we cannot easily dispatch an empty page here + */ + b+=blen+ft2len+2; + for (a=anc_buff+a_length ; b < a ; b+=ft2len+2) + insert_dynamic(info->ft1_to_ft2, b); + + /* fixing the page's length - it contains only one key now */ + anc_page->size= share->keypage_header + blen + ft2len + 2; + page_store_size(share, anc_page); + } + /* the rest will be done when we're back from recursion */ + } + } + else + { + if (share->now_transactional && + _ma_log_add(anc_page, org_anc_length, + key_pos, s_temp.changed_length, t_length, 1, + KEY_OP_DEBUG_LOG_ADD_1)) + DBUG_RETURN(-1); + } + DBUG_RETURN(0); /* There is room on page */ + } + /* Page is full */ + if (nod_flag) + insert_last=0; + /* + TODO: + Remove 'born_transactional' here. + The only reason for having it here is that the current + _ma_balance_page_ can't handle variable length keys. + */ + if (!(keyinfo->flag & (HA_VAR_LENGTH_KEY | HA_BINARY_PACK_KEY)) && + father_page && !insert_last && !info->quick_mode && + !info->s->base.born_transactional) + { + s_temp.key_pos= key_pos; + page_mark_changed(info, father_page); + DBUG_RETURN(_ma_balance_page(info, keyinfo, key, anc_page, + father_page, father_key_pos, + &s_temp)); + } + DBUG_RETURN(_ma_split_page(info, key, anc_page, + MY_MIN(org_anc_length, + info->s->max_index_block_size), + key_pos, s_temp.changed_length, t_length, + key_buff, insert_last)); +} /* _ma_insert */ + + +/** + @brief split a full page in two and assign emerging item to key + + @fn _ma_split_page() + info Maria handler + keyinfo Key handler + key Buffer for middle key + split_page Page that should be split + org_split_length Original length of split_page before key was inserted + inserted_key_pos Address in buffer where key was inserted + changed_length Number of bytes changed at 'inserted_key_pos' + move_length Number of bytes buffer was moved when key was inserted + key_buff Key buffer to use for temporary storage of key + insert_last_key If we are insert key on rightmost key page + + @note + split_buff is not stored on disk (caller has to do this) + + @return + @retval 2 ok (Middle key up from _ma_insert()) + @retval -1 error +*/ + +int _ma_split_page(MARIA_HA *info, MARIA_KEY *key, MARIA_PAGE *split_page, + uint org_split_length, + uchar *inserted_key_pos, uint changed_length, + int move_length, + uchar *key_buff, my_bool insert_last_key) +{ + uint keynr; + uint length,a_length,key_ref_length,t_length,nod_flag,key_length; + uint page_length, split_length, page_flag; + uchar *key_pos, *pos, *UNINIT_VAR(after_key); + MARIA_KEY_PARAM s_temp; + MARIA_PINNED_PAGE tmp_page_link, *page_link= &tmp_page_link; + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *keyinfo= key->keyinfo; + MARIA_KEY tmp_key; + MARIA_PAGE new_page; + int res; + DBUG_ENTER("_ma_split_page"); + + DBUG_DUMP("buff", split_page->buff, split_page->size); + + info->page_changed=1; /* Info->buff is used */ + info->keyread_buff_used=1; + page_flag= split_page->flag; + nod_flag= split_page->node; + key_ref_length= share->keypage_header + nod_flag; + + new_page.info= info; + new_page.buff= info->buff; + new_page.keyinfo= keyinfo; + + tmp_key.data= key_buff; + tmp_key.keyinfo= keyinfo; + if (insert_last_key) + key_pos= _ma_find_last_pos(&tmp_key, split_page, &after_key); + else + key_pos= _ma_find_half_pos(&tmp_key, split_page, &after_key); + if (!key_pos) + DBUG_RETURN(-1); + + key_length= tmp_key.data_length + tmp_key.ref_length; + split_length= (uint) (key_pos - split_page->buff); + a_length= split_page->size; + split_page->size= split_length; + page_store_size(share, split_page); + + key_pos=after_key; + if (nod_flag) + { + DBUG_PRINT("test",("Splitting nod")); + pos=key_pos-nod_flag; + memcpy(new_page.buff + share->keypage_header, pos, (size_t) nod_flag); + } + + /* Move middle item to key and pointer to new page */ + if ((new_page.pos= _ma_new(info, PAGECACHE_PRIORITY_HIGH, &page_link)) == + HA_OFFSET_ERROR) + DBUG_RETURN(-1); + + _ma_copy_key(key, &tmp_key); + _ma_kpointer(info, key->data + key_length, new_page.pos); + + /* Store new page */ + if (!(*keyinfo->get_key)(&tmp_key, page_flag, nod_flag, &key_pos)) + DBUG_RETURN(-1); + + t_length=(*keyinfo->pack_key)(&tmp_key, nod_flag, (uchar *) 0, + (uchar*) 0, (uchar*) 0, &s_temp); + length=(uint) ((split_page->buff + a_length) - key_pos); + memcpy(new_page.buff + key_ref_length + t_length, key_pos, + (size_t) length); + (*keyinfo->store_key)(keyinfo,new_page.buff+key_ref_length,&s_temp); + page_length= length + t_length + key_ref_length; + + bzero(new_page.buff, share->keypage_header); + /* Copy KEYFLAG_FLAG_ISNODE and KEYPAGE_FLAG_HAS_TRANSID from parent page */ + new_page.flag= page_flag; + new_page.size= page_length; + page_store_info(share, &new_page); + + /* Copy key number */ + keynr= _ma_get_keynr(share, split_page->buff); + _ma_store_keynr(share, new_page.buff, keynr); + + res= 2; /* Middle key up */ + if (share->now_transactional && _ma_log_new(&new_page, 0)) + res= -1; + + /* + Clear unitialized part of page to avoid valgrind/purify warnings + and to get a clean page that is easier to compress and compare with + pages generated with redo + */ + bzero(new_page.buff + page_length, share->block_size - page_length); + + if (_ma_write_keypage(&new_page, page_link->write_lock, + DFLT_INIT_HITS)) + res= -1; + + /* Save changes to split pages */ + if (share->now_transactional && + _ma_log_split(split_page, org_split_length, split_length, + inserted_key_pos, changed_length, move_length, + KEY_OP_NONE, (uchar*) 0, 0, 0)) + res= -1; + + DBUG_DUMP_KEY("middle_key", key); + DBUG_RETURN(res); +} /* _ma_split_page */ + + +/* + Calculate how to much to move to split a page in two + + Returns pointer to start of key. + key will contain the key. + after_key will contain the position to where the next key starts +*/ + +uchar *_ma_find_half_pos(MARIA_KEY *key, MARIA_PAGE *ma_page, + uchar **after_key) +{ + uint keys, length, key_ref_length, page_flag, nod_flag; + uchar *page, *end, *lastpos; + MARIA_HA *info= ma_page->info; + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *keyinfo= key->keyinfo; + DBUG_ENTER("_ma_find_half_pos"); + + nod_flag= ma_page->node; + key_ref_length= share->keypage_header + nod_flag; + page_flag= ma_page->flag; + length= ma_page->size - key_ref_length; + page= ma_page->buff+ key_ref_length; /* Point to first key */ + + if (!(keyinfo->flag & + (HA_PACK_KEY | HA_SPACE_PACK_USED | HA_VAR_LENGTH_KEY | + HA_BINARY_PACK_KEY)) && !(page_flag & KEYPAGE_FLAG_HAS_TRANSID)) + { + key_ref_length= keyinfo->keylength+nod_flag; + key->data_length= keyinfo->keylength - info->s->rec_reflength; + key->ref_length= info->s->rec_reflength; + key->flag= 0; + keys=length/(key_ref_length*2); + end=page+keys*key_ref_length; + *after_key=end+key_ref_length; + memcpy(key->data, end, key_ref_length); + DBUG_RETURN(end); + } + + end=page+length/2-key_ref_length; /* This is aprox. half */ + key->data[0]= 0; /* Safety */ + do + { + lastpos=page; + if (!(length= (*keyinfo->get_key)(key, page_flag, nod_flag, &page))) + DBUG_RETURN(0); + } while (page < end); + *after_key= page; + DBUG_PRINT("exit",("returns: %p page: %p half: %p", + lastpos, page, end)); + DBUG_RETURN(lastpos); +} /* _ma_find_half_pos */ + + +/** + Find second to last key on leaf page + + @notes + Used to split buffer at last key. In this case the next to last + key will be moved to parent page and last key will be on it's own page. + + @TODO + Add one argument for 'last key value' to get_key so that one can + do the loop without having to copy the found key the whole time + + @return + @retval Pointer to the start of the key before the last key + @retval int_key will contain the last key +*/ + +static uchar *_ma_find_last_pos(MARIA_KEY *int_key, MARIA_PAGE *ma_page, + uchar **after_key) +{ + uint keys, length, key_ref_length, page_flag; + uchar *page, *end, *lastpos, *prevpos; + uchar key_buff[MARIA_MAX_KEY_BUFF]; + MARIA_HA *info= ma_page->info; + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *keyinfo= int_key->keyinfo; + MARIA_KEY tmp_key; + DBUG_ENTER("_ma_find_last_pos"); + + key_ref_length= share->keypage_header; + page_flag= ma_page->flag; + length= ma_page->size - key_ref_length; + page= ma_page->buff + key_ref_length; + + if (!(keyinfo->flag & + (HA_PACK_KEY | HA_SPACE_PACK_USED | HA_VAR_LENGTH_KEY | + HA_BINARY_PACK_KEY)) && !(page_flag & KEYPAGE_FLAG_HAS_TRANSID)) + { + keys= length / keyinfo->keylength - 2; + length= keyinfo->keylength; + int_key->data_length= length - info->s->rec_reflength; + int_key->ref_length= info->s->rec_reflength; + int_key->flag= 0; + end=page+keys*length; + *after_key=end+length; + memcpy(int_key->data, end, length); + DBUG_RETURN(end); + } + + end=page+length-key_ref_length; + lastpos=page; + tmp_key.data= key_buff; + tmp_key.keyinfo= int_key->keyinfo; + key_buff[0]= 0; /* Safety */ + + /* We know that there are at least 2 keys on the page */ + + if (!(length=(*keyinfo->get_key)(&tmp_key, page_flag, 0, &page))) + { + _ma_set_fatal_error(info, HA_ERR_CRASHED); + DBUG_RETURN(0); + } + + do + { + prevpos=lastpos; lastpos=page; + int_key->data_length= tmp_key.data_length; + int_key->ref_length= tmp_key.ref_length; + int_key->flag= tmp_key.flag; + memcpy(int_key->data, key_buff, length); /* previous key */ + if (!(length=(*keyinfo->get_key)(&tmp_key, page_flag, 0, &page))) + { + _ma_set_fatal_error(info, HA_ERR_CRASHED); + DBUG_RETURN(0); + } + } while (page < end); + + *after_key=lastpos; + DBUG_PRINT("exit",("returns: %p page: %p end: %p", + prevpos,page,end)); + DBUG_RETURN(prevpos); +} /* _ma_find_last_pos */ + + +/** + @brief Balance page with static size keys with page on right/left + + @param key Middle key will be stored here + + @notes + Father_buff will always be changed + Caller must handle saving of curr_buff + + @return + @retval 0 Balance was done (father buff is saved) + @retval 1 Middle key up (father buff is not saved) + @retval -1 Error +*/ + +static int _ma_balance_page(MARIA_HA *info, MARIA_KEYDEF *keyinfo, + MARIA_KEY *key, MARIA_PAGE *curr_page, + MARIA_PAGE *father_page, + uchar *father_key_pos, MARIA_KEY_PARAM *s_temp) +{ + MARIA_PINNED_PAGE tmp_page_link, *new_page_link= &tmp_page_link; + MARIA_SHARE *share= info->s; + my_bool right, buff_alloced; + uint k_length,father_length,father_keylength,nod_flag,curr_keylength; + uint right_length,left_length,new_right_length,new_left_length,extra_length; + uint keys, tmp_length, extra_buff_length; + uchar *pos, *extra_buff, *parting_key; + uchar *tmp_part_key; + MARIA_PAGE next_page, extra_page, *left_page, *right_page; + DBUG_ENTER("_ma_balance_page"); + + alloc_on_stack(*info->stack_end_ptr, tmp_part_key, buff_alloced, + keyinfo->max_store_length); + if (!tmp_part_key) + DBUG_RETURN(-1); + + k_length= keyinfo->keylength; + father_length= father_page->size; + father_keylength= k_length + share->base.key_reflength; + nod_flag= curr_page->node; + curr_keylength= k_length+nod_flag; + info->page_changed=1; + + if ((father_key_pos != father_page->buff+father_length && + (info->state->records & 1)) || + father_key_pos == father_page->buff+ share->keypage_header + + share->base.key_reflength) + { + right=1; + next_page.pos= _ma_kpos(share->base.key_reflength, + father_key_pos+father_keylength); + left_page= curr_page; + right_page= &next_page; + DBUG_PRINT("info", ("use right page: %lu", + (ulong) (next_page.pos / keyinfo->block_length))); + } + else + { + right=0; + father_key_pos-=father_keylength; + next_page.pos= _ma_kpos(share->base.key_reflength,father_key_pos); + left_page= &next_page; + right_page= curr_page; + DBUG_PRINT("info", ("use left page: %lu", + (ulong) (next_page.pos / keyinfo->block_length))); + } /* father_key_pos ptr to parting key */ + + if (_ma_fetch_keypage(&next_page, info, keyinfo, next_page.pos, + PAGECACHE_LOCK_WRITE, + DFLT_INIT_HITS, info->buff, 0)) + goto err; + page_mark_changed(info, &next_page); + DBUG_DUMP("next", next_page.buff, next_page.size); + + /* Test if there is room to share keys */ + left_length= left_page->size; + right_length= right_page->size; + keys= ((left_length+right_length-share->keypage_header*2-nod_flag*2)/ + curr_keylength); + + if ((right ? right_length : left_length) + curr_keylength <= + share->max_index_block_size) + { + /* Enough space to hold all keys in the two buffers ; Balance bufferts */ + new_left_length= share->keypage_header+nod_flag+(keys/2)*curr_keylength; + new_right_length=share->keypage_header+nod_flag+(((keys+1)/2)* + curr_keylength); + left_page->size= new_left_length; + page_store_size(share, left_page); + right_page->size= new_right_length; + page_store_size(share, right_page); + + DBUG_PRINT("info", ("left_length: %u -> %u right_length: %u -> %u", + left_length, new_left_length, + right_length, new_right_length)); + if (left_length < new_left_length) + { + uint length; + DBUG_PRINT("info", ("move keys to end of buff")); + + /* Move keys right_page -> left_page */ + pos= left_page->buff+left_length; + memcpy(pos,father_key_pos, (size_t) k_length); + memcpy(pos+k_length, right_page->buff + share->keypage_header, + (size_t) (length=new_left_length - left_length - k_length)); + pos= right_page->buff + share->keypage_header + length; + memcpy(father_key_pos, pos, (size_t) k_length); + bmove(right_page->buff + share->keypage_header, + pos + k_length, new_right_length - share->keypage_header); + + if (share->now_transactional) + { + if (right) + { + /* + Log changes to page on left + The original page is on the left and stored in left_page->buff + We have on the page the newly inserted key and data + from buff added last on the page + */ + if (_ma_log_split(curr_page, + left_length - s_temp->move_length, + new_left_length, + s_temp->key_pos, s_temp->changed_length, + s_temp->move_length, + KEY_OP_ADD_SUFFIX, + curr_page->buff + left_length, + new_left_length - left_length, + new_left_length - left_length+ k_length)) + goto err; + /* + Log changes to page on right + This contains the original data with some keys deleted from + start of page + */ + if (_ma_log_prefix(&next_page, 0, + ((int) new_right_length - (int) right_length), + KEY_OP_DEBUG_LOG_PREFIX_3)) + goto err; + } + else + { + /* + Log changes to page on right (the original page) which is in buff + Data is removed from start of page + The inserted key may be in buff or moved to curr_buff + */ + if (_ma_log_del_prefix(curr_page, + right_length - s_temp->changed_length, + new_right_length, + s_temp->key_pos, s_temp->changed_length, + s_temp->move_length)) + goto err; + /* + Log changes to page on left, which has new data added last + */ + if (_ma_log_suffix(&next_page, left_length, new_left_length)) + goto err; + } + } + } + else + { + uint length; + DBUG_PRINT("info", ("move keys to start of right_page")); + + bmove_upp(right_page->buff + new_right_length, + right_page->buff + right_length, + right_length - share->keypage_header); + length= new_right_length -right_length - k_length; + memcpy(right_page->buff + share->keypage_header + length, father_key_pos, + (size_t) k_length); + pos= left_page->buff + new_left_length; + memcpy(father_key_pos, pos, (size_t) k_length); + memcpy(right_page->buff + share->keypage_header, pos+k_length, + (size_t) length); + + if (share->now_transactional) + { + if (right) + { + /* + Log changes to page on left + The original page is on the left and stored in curr_buff + The page is shortened from end and the key may be on the page + */ + if (_ma_log_split(curr_page, + left_length - s_temp->move_length, + new_left_length, + s_temp->key_pos, s_temp->changed_length, + s_temp->move_length, + KEY_OP_NONE, (uchar*) 0, 0, 0)) + goto err; + /* + Log changes to page on right + This contains the original data, with some data from cur_buff + added first + */ + if (_ma_log_prefix(&next_page, + (uint) (new_right_length - right_length), + (int) (new_right_length - right_length), + KEY_OP_DEBUG_LOG_PREFIX_4)) + goto err; + } + else + { + /* + Log changes to page on right (the original page) which is in buff + We have on the page the newly inserted key and data + from buff added first on the page + */ + uint diff_length= new_right_length - right_length; + if (_ma_log_split(curr_page, + left_length - s_temp->move_length, + new_right_length, + s_temp->key_pos + diff_length, + s_temp->changed_length, + s_temp->move_length, + KEY_OP_ADD_PREFIX, + curr_page->buff + share->keypage_header, + diff_length, diff_length + k_length)) + goto err; + /* + Log changes to page on left, which is shortened from end + */ + if (_ma_log_suffix(&next_page, left_length, new_left_length)) + goto err; + } + } + } + + /* Log changes to father (one level up) page */ + + if (share->now_transactional && + _ma_log_change(father_page, father_key_pos, k_length, + KEY_OP_DEBUG_FATHER_CHANGED_1)) + goto err; + + /* + next_page_link->changed is marked as true above and fathers + page_link->changed is marked as true in caller + */ + if (_ma_write_keypage(&next_page, PAGECACHE_LOCK_LEFT_WRITELOCKED, + DFLT_INIT_HITS) || + _ma_write_keypage(father_page, + PAGECACHE_LOCK_LEFT_WRITELOCKED, DFLT_INIT_HITS)) + goto err; + stack_alloc_free(tmp_part_key, buff_alloced); + DBUG_RETURN(0); + } + + /* left_page and right_page are full, lets split and make new nod */ + + extra_buff= info->buff+share->base.max_key_block_length; + new_left_length= new_right_length= (share->keypage_header + nod_flag + + (keys+1) / 3 * curr_keylength); + extra_page.info= info; + extra_page.keyinfo= keyinfo; + extra_page.buff= extra_buff; + + /* + 5 is the minum number of keys we can have here. This comes from + the fact that each full page can store at least 2 keys and in this case + we have a 'split' key, ie 2+2+1 = 5 + */ + if (keys == 5) /* Too few keys to balance */ + new_left_length-=curr_keylength; + extra_length= (nod_flag + left_length + right_length - + new_left_length - new_right_length - curr_keylength); + extra_buff_length= extra_length + share->keypage_header; + DBUG_PRINT("info",("left_length: %d right_length: %d new_left_length: %d new_right_length: %d extra_length: %d", + left_length, right_length, + new_left_length, new_right_length, + extra_length)); + + left_page->size= new_left_length; + page_store_size(share, left_page); + right_page->size= new_right_length; + page_store_size(share, right_page); + + bzero(extra_buff, share->keypage_header); + extra_page.flag= nod_flag ? KEYPAGE_FLAG_ISNOD : 0; + extra_page.size= extra_buff_length; + page_store_info(share, &extra_page); + + /* Copy key number */ + _ma_store_keynr(share, extra_buff, keyinfo->key_nr); + + /* move first largest keys to new page */ + pos= right_page->buff + right_length-extra_length; + memcpy(extra_buff + share->keypage_header, pos, extra_length); + /* Zero old data from buffer */ + bzero(extra_buff + extra_buff_length, + share->block_size - extra_buff_length); + + /* Save new parting key between buff and extra_buff */ + memcpy(tmp_part_key, pos-k_length,k_length); + /* Make place for new keys */ + bmove_upp(right_page->buff + new_right_length, pos - k_length, + right_length - extra_length - k_length - share->keypage_header); + /* Copy keys from left page */ + pos= left_page->buff + new_left_length; + memcpy(right_page->buff + share->keypage_header, pos + k_length, + (size_t) (tmp_length= left_length - new_left_length - k_length)); + /* Copy old parting key */ + parting_key= right_page->buff + share->keypage_header + tmp_length; + memcpy(parting_key, father_key_pos, (size_t) k_length); + + /* Move new parting keys up to caller */ + memcpy((right ? key->data : father_key_pos),pos,(size_t) k_length); + memcpy((right ? father_key_pos : key->data),tmp_part_key, k_length); + + if ((extra_page.pos= _ma_new(info, DFLT_INIT_HITS, &new_page_link)) + == HA_OFFSET_ERROR) + goto err; + _ma_kpointer(info,key->data+k_length, extra_page.pos); + /* This is safe as long we are using not keys with transid */ + key->data_length= k_length - info->s->rec_reflength; + key->ref_length= info->s->rec_reflength; + + if (right) + { + /* + Page order according to key values: + orignal_page (curr_page = left_page), next_page (buff), extra_buff + + Move page positions so that we store data in extra_page where + next_page was and next_page will be stored at the new position + */ + swap_variables(my_off_t, extra_page.pos, next_page.pos); + } + + if (share->now_transactional) + { + if (right) + { + /* + left_page is shortened, + right_page is getting new keys at start and shortened from end. + extra_page is new page + + Note that extra_page (largest key parts) will be stored at the + place of the original 'right' page (next_page) and right page + will be stored at the new page position + + This makes the log entries smaller as right_page contains all + data to generate the data extra_buff + */ + + /* + Log changes to page on left (page shortened page at end) + */ + if (_ma_log_split(curr_page, + left_length - s_temp->move_length, new_left_length, + s_temp->key_pos, s_temp->changed_length, + s_temp->move_length, + KEY_OP_NONE, (uchar*) 0, 0, 0)) + goto err; + /* + Log changes to right page (stored at next page) + This contains the last 'extra_buff' from 'buff' + */ + if (_ma_log_prefix(&extra_page, + 0, (int) (extra_buff_length - right_length), + KEY_OP_DEBUG_LOG_PREFIX_5)) + goto err; + + /* + Log changes to middle page, which is stored at the new page + position + */ + if (_ma_log_new(&next_page, 0)) + goto err; + } + else + { + /* + Log changes to page on right (the original page) which is in buff + This contains the original data, with some data from curr_buff + added first and shortened at end + */ + int data_added_first= left_length - new_left_length; + if (_ma_log_key_middle(right_page, + new_right_length, + data_added_first, + data_added_first, + extra_length, + s_temp->key_pos, + s_temp->changed_length, + s_temp->move_length)) + goto err; + + /* Log changes to page on left, which is shortened from end */ + if (_ma_log_suffix(left_page, left_length, new_left_length)) + goto err; + + /* Log change to rightmost (new) page */ + if (_ma_log_new(&extra_page, 0)) + goto err; + } + + /* Log changes to father (one level up) page */ + if (share->now_transactional && + _ma_log_change(father_page, father_key_pos, k_length, + KEY_OP_DEBUG_FATHER_CHANGED_2)) + goto err; + } + + if (_ma_write_keypage(&next_page, + (right ? new_page_link->write_lock : + PAGECACHE_LOCK_LEFT_WRITELOCKED), + DFLT_INIT_HITS) || + _ma_write_keypage(&extra_page, + (!right ? new_page_link->write_lock : + PAGECACHE_LOCK_LEFT_WRITELOCKED), + DFLT_INIT_HITS)) + goto err; + + stack_alloc_free(tmp_part_key, buff_alloced); + DBUG_RETURN(1); /* Middle key up */ + +err: + stack_alloc_free(tmp_part_key, buff_alloced); + DBUG_RETURN(-1); +} /* _ma_balance_page */ + + +/********************************************************************** + * Bulk insert code * + **********************************************************************/ + +typedef struct { + MARIA_HA *info; + uint keynr; +} bulk_insert_param; + + +static my_bool _ma_ck_write_tree(register MARIA_HA *info, MARIA_KEY *key) +{ + my_bool error; + uint keynr= key->keyinfo->key_nr; + DBUG_ENTER("_ma_ck_write_tree"); + + /* Store ref_length as this is always constant */ + info->bulk_insert_ref_length= key->ref_length; + error= tree_insert(&info->bulk_insert[keynr], key->data, + key->data_length + key->ref_length, + info->bulk_insert[keynr].custom_arg) == 0; + DBUG_RETURN(error); +} /* _ma_ck_write_tree */ + + +/* typeof(_ma_keys_compare)=qsort_cmp2 */ + +static int keys_compare(bulk_insert_param *param, uchar *key1, uchar *key2) +{ + uint not_used[2]; + return ha_key_cmp(param->info->s->keyinfo[param->keynr].seg, + key1, key2, USE_WHOLE_KEY, SEARCH_SAME, + not_used); +} + + +static int keys_free(void* key_arg, TREE_FREE mode, void *param_arg) +{ + /* + Probably I can use info->lastkey here, but I'm not sure, + and to be safe I'd better use local lastkey. + */ + bulk_insert_param *param= (bulk_insert_param*)param_arg; + MARIA_SHARE *share= param->info->s; + uchar lastkey[MARIA_MAX_KEY_BUFF], *key= (uchar*)key_arg; + uint keylen; + MARIA_KEYDEF *keyinfo= share->keyinfo + param->keynr; + MARIA_KEY tmp_key; + + switch (mode) { + case free_init: + if (share->lock_key_trees) + { + mysql_rwlock_wrlock(&keyinfo->root_lock); + keyinfo->version++; + } + return 0; + case free_free: + /* Note: keylen doesn't contain transid lengths */ + keylen= _ma_keylength(keyinfo, key); + tmp_key.data= lastkey; + tmp_key.keyinfo= keyinfo; + tmp_key.data_length= keylen - share->rec_reflength; + tmp_key.ref_length= param->info->bulk_insert_ref_length; + tmp_key.flag= (param->info->bulk_insert_ref_length == + share->rec_reflength ? 0 : SEARCH_USER_KEY_HAS_TRANSID); + /* + We have to copy key as ma_ck_write_btree may need the buffer for + copying middle key up if tree is growing + */ + memcpy(lastkey, key, tmp_key.data_length + tmp_key.ref_length); + _ma_ck_write_btree(param->info, &tmp_key); + return 0; + case free_end: + if (share->lock_key_trees) + mysql_rwlock_unlock(&keyinfo->root_lock); + return 0; + } + return 0; +} + + +int maria_init_bulk_insert(MARIA_HA *info, size_t cache_size, ha_rows rows) +{ + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *key=share->keyinfo; + bulk_insert_param *params; + uint i, num_keys, total_keylength; + ulonglong key_map; + DBUG_ENTER("_ma_init_bulk_insert"); + DBUG_PRINT("enter",("cache_size: %lu", (ulong) cache_size)); + + DBUG_ASSERT(!info->bulk_insert && + (!rows || rows >= MARIA_MIN_ROWS_TO_USE_BULK_INSERT)); + + maria_clear_all_keys_active(key_map); + for (i=total_keylength=num_keys=0 ; i < share->base.keys ; i++) + { + if (! (key[i].flag & HA_NOSAME) && (share->base.auto_key != i + 1) && + maria_is_key_active(share->state.key_map, i)) + { + num_keys++; + maria_set_key_active(key_map, i); + total_keylength+=key[i].maxlength+TREE_ELEMENT_EXTRA_SIZE; + } + } + + if (num_keys==0 || + num_keys * (size_t) MARIA_MIN_SIZE_BULK_INSERT_TREE > cache_size) + DBUG_RETURN(0); + + if (rows && rows*total_keylength < cache_size) + cache_size= (size_t)rows; + else + cache_size/=total_keylength*16; + + info->bulk_insert=(TREE *) + my_malloc(PSI_INSTRUMENT_ME, (sizeof(TREE)*share->base.keys+ + sizeof(bulk_insert_param)*num_keys),MYF(0)); + + if (!info->bulk_insert) + DBUG_RETURN(HA_ERR_OUT_OF_MEM); + + params=(bulk_insert_param *)(info->bulk_insert+share->base.keys); + for (i=0 ; i < share->base.keys ; i++) + { + if (maria_is_key_active(key_map, i)) + { + params->info=info; + params->keynr=i; + /* Only allocate a 16'th of the buffer at a time */ + init_tree(&info->bulk_insert[i], + cache_size * key[i].maxlength, + cache_size * key[i].maxlength, 0, + (qsort_cmp2) keys_compare, keys_free, (void *)params++, MYF(0)); + } + else + info->bulk_insert[i].root=0; + } + + DBUG_RETURN(0); +} + +void maria_flush_bulk_insert(MARIA_HA *info, uint inx) +{ + if (info->bulk_insert) + { + if (is_tree_inited(&info->bulk_insert[inx])) + reset_tree(&info->bulk_insert[inx]); + } +} + + +int maria_end_bulk_insert(MARIA_HA *info, my_bool abort) +{ + int first_error= 0; + DBUG_ENTER("maria_end_bulk_insert"); + if (info->bulk_insert) + { + uint i; + for (i=0 ; i < info->s->base.keys ; i++) + { + if (is_tree_inited(&info->bulk_insert[i])) + { + int error; + if (info->s->deleting) + reset_free_element(&info->bulk_insert[i]); + if ((error= delete_tree(&info->bulk_insert[i], abort))) + { + first_error= first_error ? first_error : error; + abort= 1; + } + } + } + my_free(info->bulk_insert); + info->bulk_insert= 0; + } + DBUG_RETURN(first_error); +} + + +/**************************************************************************** + Dedicated functions that generate log entries +****************************************************************************/ + + +int _ma_write_undo_key_insert(MARIA_HA *info, const MARIA_KEY *key, + my_off_t *root, my_off_t new_root, LSN *res_lsn) +{ + MARIA_SHARE *share= info->s; + MARIA_KEYDEF *keyinfo= key->keyinfo; + uchar log_data[LSN_STORE_SIZE + FILEID_STORE_SIZE + + KEY_NR_STORE_SIZE]; + const uchar *key_value; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 2]; + struct st_msg_to_write_hook_for_undo_key msg; + uint key_length; + + /* Save if we need to write a clr record */ + lsn_store(log_data, info->trn->undo_lsn); + key_nr_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE, + keyinfo->key_nr); + key_length= key->data_length + key->ref_length; + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data); + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= key->data; + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= key_length; + + msg.root= root; + msg.value= new_root; + msg.auto_increment= 0; + key_value= key->data; + if (share->base.auto_key == ((uint) keyinfo->key_nr + 1)) + { + const HA_KEYSEG *keyseg= keyinfo->seg; + uchar reversed[MARIA_MAX_KEY_BUFF]; + if (keyseg->flag & HA_SWAP_KEY) + { + /* We put key from log record to "data record" packing format... */ + const uchar *key_ptr= key->data, *key_end= key->data + keyseg->length; + uchar *to= reversed + keyseg->length; + do + { + *--to= *key_ptr++; + } while (key_ptr != key_end); + key_value= to; + } + /* ... so that we can read it with: */ + msg.auto_increment= + ma_retrieve_auto_increment(key_value, keyseg->type); + /* and write_hook_for_undo_key_insert() will pick this. */ + } + + return translog_write_record(res_lsn, LOGREC_UNDO_KEY_INSERT, + info->trn, info, + (translog_size_t) + log_array[TRANSLOG_INTERNAL_PARTS + 0].length + + key_length, + TRANSLOG_INTERNAL_PARTS + 2, log_array, + log_data + LSN_STORE_SIZE, &msg) ? -1 : 0; +} + + +/** + @brief Log creation of new page + + @note + We don't have to store the page_length into the log entry as we can + calculate this from the length of the log entry + + @retval 1 error + @retval 0 ok +*/ + +my_bool _ma_log_new(MARIA_PAGE *ma_page, my_bool root_page) +{ + LSN lsn; + uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE * 2 + KEY_NR_STORE_SIZE + +1]; + uint page_length; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 2]; + MARIA_HA *info= ma_page->info; + MARIA_SHARE *share= info->s; + my_off_t page= ma_page->pos / share->block_size; + DBUG_ENTER("_ma_log_new"); + DBUG_PRINT("enter", ("page: %lu", (ulong) page)); + + DBUG_ASSERT(share->now_transactional); + + /* Store address of new root page */ + page_store(log_data + FILEID_STORE_SIZE, page); + + /* Store link to next unused page */ + if (info->key_del_used == 2) + page= 0; /* key_del not changed */ + else + page= ((share->key_del_current == HA_OFFSET_ERROR) ? IMPOSSIBLE_PAGE_NO : + share->key_del_current / share->block_size); + + page_store(log_data + FILEID_STORE_SIZE + PAGE_STORE_SIZE, page); + key_nr_store(log_data + FILEID_STORE_SIZE + PAGE_STORE_SIZE*2, + ma_page->keyinfo->key_nr); + log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE*2 + KEY_NR_STORE_SIZE]= + (uchar) root_page; + + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= sizeof(log_data); + + page_length= ma_page->size - LSN_STORE_SIZE; + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= ma_page->buff + LSN_STORE_SIZE; + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= page_length; + + /* Remember new page length for future log entires for same page */ + ma_page->org_size= ma_page->size; + + if (translog_write_record(&lsn, LOGREC_REDO_INDEX_NEW_PAGE, + info->trn, info, + (translog_size_t) + (sizeof(log_data) + page_length), + TRANSLOG_INTERNAL_PARTS + 2, log_array, + log_data, NULL)) + DBUG_RETURN(1); + DBUG_RETURN(0); +} + + +/** + @brief + Log when some part of the key page changes +*/ + +my_bool _ma_log_change(MARIA_PAGE *ma_page, const uchar *key_pos, uint length, + enum en_key_debug debug_marker __attribute__((unused))) +{ + LSN lsn; + uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 2 + 6 + 7], *log_pos; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 4]; + uint offset= (uint) (key_pos - ma_page->buff), translog_parts; + MARIA_HA *info= ma_page->info; + my_off_t page= ma_page->pos / info->s->block_size; + DBUG_ENTER("_ma_log_change"); + DBUG_PRINT("enter", ("page: %lu length: %u", (ulong) page, length)); + + DBUG_ASSERT(info->s->now_transactional); + DBUG_ASSERT(offset + length <= ma_page->size); + DBUG_ASSERT(ma_page->org_size == ma_page->size); + + /* Store address of new root page */ + page= ma_page->pos / info->s->block_size; + page_store(log_data + FILEID_STORE_SIZE, page); + log_pos= log_data+ FILEID_STORE_SIZE + PAGE_STORE_SIZE; + +#ifdef EXTRA_DEBUG_KEY_CHANGES + (*log_pos++)= KEY_OP_DEBUG; + (*log_pos++)= debug_marker; +#endif + + log_pos[0]= KEY_OP_OFFSET; + int2store(log_pos+1, offset); + log_pos[3]= KEY_OP_CHANGE; + int2store(log_pos+4, length); + log_pos+= 6; + + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (log_pos - log_data); + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= key_pos; + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= length; + translog_parts= 2; + + _ma_log_key_changes(ma_page, + log_array + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_pos, &length, &translog_parts); + + if (translog_write_record(&lsn, LOGREC_REDO_INDEX, + info->trn, info, + (translog_size_t) (log_pos - log_data) + length, + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_array, log_data, NULL)) + DBUG_RETURN(1); + DBUG_RETURN(0); +} + + +/** + @brief Write log entry for page splitting + + @fn _ma_log_split() + @param + ma_page Page that is changed + org_length Original length of page. Can be bigger than block_size + for block that overflowed + new_length New length of page + key_pos Where key is inserted on page (may be 0 if no key) + key_length Number of bytes changed at key_pos + move_length Number of bytes moved at key_pos to make room for key + prefix_or_suffix KEY_OP_NONE Ignored + KEY_OP_ADD_PREFIX Add data to start of page + KEY_OP_ADD_SUFFIX Add data to end of page + data What data was added + data_length Number of bytes added first or last + changed_length Number of bytes changed first or last. + + @note + Write log entry for page that has got a key added to the page under + one and only one of the following senarios: + - Page is shortened from end + - Data is added to end of page + - Data added at front of page +*/ + +static my_bool _ma_log_split(MARIA_PAGE *ma_page, + uint org_length, uint new_length, + const uchar *key_pos, uint key_length, + int move_length, enum en_key_op prefix_or_suffix, + const uchar *data, uint data_length, + uint changed_length) +{ + LSN lsn; + uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 2 + 2 + 3+3+3+3+3+2 +7]; + uchar *log_pos; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 6]; + uint offset= (uint) (key_pos - ma_page->buff); + uint translog_parts, extra_length; + MARIA_HA *info= ma_page->info; + my_off_t page= ma_page->pos / info->s->block_size; + DBUG_ENTER("_ma_log_split"); + DBUG_PRINT("enter", ("page: %lu org_length: %u new_length: %u", + (ulong) page, org_length, new_length)); + + DBUG_ASSERT(changed_length >= data_length); + DBUG_ASSERT(org_length <= info->s->max_index_block_size); + DBUG_ASSERT(new_length == ma_page->size); + DBUG_ASSERT(org_length == ma_page->org_size); + + log_pos= log_data + FILEID_STORE_SIZE; + page_store(log_pos, page); + log_pos+= PAGE_STORE_SIZE; + +#ifdef EXTRA_DEBUG_KEY_CHANGES + (*log_pos++)= KEY_OP_DEBUG; + (*log_pos++)= KEY_OP_DEBUG_LOG_SPLIT; +#endif + + /* Store keypage_flag */ + *log_pos++= KEY_OP_SET_PAGEFLAG; + *log_pos++= _ma_get_keypage_flag(info->s, ma_page->buff); + + if (new_length <= offset || !key_pos) + { + /* + Page was split before inserted key. Write redo entry where + we just cut current page at page_length + */ + uint length_offset= org_length - new_length; + log_pos[0]= KEY_OP_DEL_SUFFIX; + int2store(log_pos+1, length_offset); + log_pos+= 3; + translog_parts= 1; + extra_length= 0; + DBUG_ASSERT(data_length == 0); + } + else + { + /* Key was added to page which was split after the inserted key */ + uint max_key_length; + + /* + Handle case when split happened directly after the newly inserted key. + */ + max_key_length= new_length - offset; + extra_length= MY_MIN(key_length, max_key_length); + if (offset + move_length > new_length) + { + /* This is true when move_length includes changes for next packed key */ + move_length= new_length - offset; + } + + if ((int) new_length < (int) (org_length + move_length + data_length)) + { + /* Shorten page */ + uint diff= org_length + move_length + data_length - new_length; + log_pos[0]= KEY_OP_DEL_SUFFIX; + int2store(log_pos + 1, diff); + log_pos+= 3; + DBUG_ASSERT(data_length == 0); /* Page is shortened */ + DBUG_ASSERT(offset <= org_length - diff); + } + else + { + DBUG_ASSERT(new_length == org_length + move_length + data_length); + DBUG_ASSERT(offset <= org_length); + } + + log_pos[0]= KEY_OP_OFFSET; + int2store(log_pos+1, offset); + log_pos+= 3; + + if (move_length) + { + log_pos[0]= KEY_OP_SHIFT; + int2store(log_pos+1, move_length); + log_pos+= 3; + } + + log_pos[0]= KEY_OP_CHANGE; + int2store(log_pos+1, extra_length); + log_pos+= 3; + + /* Point to original inserted key data */ + if (prefix_or_suffix == KEY_OP_ADD_PREFIX) + key_pos+= data_length; + + translog_parts= 2; + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= key_pos; + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= extra_length; + } + + if (data_length) + { + /* Add prefix or suffix */ + log_pos[0]= prefix_or_suffix; + int2store(log_pos+1, data_length); + log_pos+= 3; + if (prefix_or_suffix == KEY_OP_ADD_PREFIX) + { + int2store(log_pos+1, changed_length); + log_pos+= 2; + data_length= changed_length; + } + log_array[TRANSLOG_INTERNAL_PARTS + translog_parts].str= data; + log_array[TRANSLOG_INTERNAL_PARTS + translog_parts].length= data_length; + translog_parts++; + extra_length+= data_length; + } + + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - + log_data); + + _ma_log_key_changes(ma_page, + log_array + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_pos, &extra_length, &translog_parts); + /* Remember new page length for future log entires for same page */ + ma_page->org_size= ma_page->size; + + DBUG_RETURN(translog_write_record(&lsn, LOGREC_REDO_INDEX, + info->trn, info, + (translog_size_t) + log_array[TRANSLOG_INTERNAL_PARTS + + 0].length + extra_length, + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_array, log_data, NULL)); +} + + +/** + @brief + Write log entry for page that has got a key added to the page + and page is shortened from start of page + + @fn _ma_log_del_prefix() + @param info Maria handler + @param page Page number + @param buff Page buffer + @param org_length Length of buffer when read + @param new_length Final length + @param key_pos Where on page buffer key was added. This is position + before prefix was removed + @param key_length How many bytes was changed at 'key_pos' + @param move_length How many bytes was moved up when key was added + + @return + @retval 0 ok + @retval 1 error +*/ + +static my_bool _ma_log_del_prefix(MARIA_PAGE *ma_page, + uint org_length, uint new_length, + const uchar *key_pos, uint key_length, + int move_length) +{ + LSN lsn; + uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 2 + 2 + 12 + 7]; + uchar *log_pos; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 4]; + uint offset= (uint) (key_pos - ma_page->buff); + uint diff_length= org_length + move_length - new_length; + uint translog_parts, extra_length; + MARIA_HA *info= ma_page->info; + my_off_t page= ma_page->pos / info->s->block_size; + DBUG_ENTER("_ma_log_del_prefix"); + DBUG_PRINT("enter", ("page: %lu org_length: %u new_length: %u", + (ulong) page, org_length, new_length)); + + DBUG_ASSERT((int) diff_length > 0); + DBUG_ASSERT(ma_page->org_size == org_length); + DBUG_ASSERT(ma_page->size == new_length); + + log_pos= log_data + FILEID_STORE_SIZE; + page_store(log_pos, page); + log_pos+= PAGE_STORE_SIZE; + + translog_parts= 1; + extra_length= 0; + +#ifdef EXTRA_DEBUG_KEY_CHANGES + *log_pos++= KEY_OP_DEBUG; + *log_pos++= KEY_OP_DEBUG_LOG_DEL_PREFIX; +#endif + + /* Store keypage_flag */ + *log_pos++= KEY_OP_SET_PAGEFLAG; + *log_pos++= _ma_get_keypage_flag(info->s, ma_page->buff); + + if (offset < diff_length + info->s->keypage_header) + { + /* + Key is not anymore on page. Move data down, but take into account that + the original page had grown with 'move_length bytes' + */ + DBUG_ASSERT(offset + key_length <= diff_length + info->s->keypage_header); + + log_pos[0]= KEY_OP_DEL_PREFIX; + int2store(log_pos+1, diff_length - move_length); + log_pos+= 3; + } + else + { + /* + Correct position to key, as data before key has been delete and key + has thus been moved down + */ + offset-= diff_length; + key_pos-= diff_length; + + /* Move data down */ + log_pos[0]= KEY_OP_DEL_PREFIX; + int2store(log_pos+1, diff_length); + log_pos+= 3; + + log_pos[0]= KEY_OP_OFFSET; + int2store(log_pos+1, offset); + log_pos+= 3; + + if (move_length) + { + log_pos[0]= KEY_OP_SHIFT; + int2store(log_pos+1, move_length); + log_pos+= 3; + } + log_pos[0]= KEY_OP_CHANGE; + int2store(log_pos+1, key_length); + log_pos+= 3; + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= key_pos; + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= key_length; + translog_parts= 2; + extra_length= key_length; + } + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - + log_data); + _ma_log_key_changes(ma_page, + log_array + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_pos, &extra_length, &translog_parts); + /* Remember new page length for future log entires for same page */ + ma_page->org_size= ma_page->size; + + DBUG_RETURN(translog_write_record(&lsn, LOGREC_REDO_INDEX, + info->trn, info, + (translog_size_t) + log_array[TRANSLOG_INTERNAL_PARTS + + 0].length + extra_length, + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_array, log_data, NULL)); +} + + +/** + @brief + Write log entry for page that has got data added first and + data deleted last. Old changed key may be part of page +*/ + +static my_bool _ma_log_key_middle(MARIA_PAGE *ma_page, + uint new_length, + uint data_added_first, + uint data_changed_first, + uint data_deleted_last, + const uchar *key_pos, + uint key_length, int move_length) +{ + LSN lsn; + uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 2 + 2 + 3+5+3+3+3 + 7]; + uchar *log_pos; + LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 6]; + uint key_offset; + uint translog_parts, extra_length; + MARIA_HA *info= ma_page->info; + my_off_t page= ma_page->pos / info->s->block_size; + DBUG_ENTER("_ma_log_key_middle"); + DBUG_PRINT("enter", ("page: %lu", (ulong) page)); + + DBUG_ASSERT(ma_page->size == new_length); + + /* new place of key after changes */ + key_pos+= data_added_first; + key_offset= (uint) (key_pos - ma_page->buff); + if (key_offset < new_length) + { + /* key is on page; Calculate how much of the key is there */ + uint max_key_length= new_length - key_offset; + if (max_key_length < key_length) + { + /* Key is last on page */ + key_length= max_key_length; + move_length= 0; + } + /* + Take into account that new data was added as part of original key + that also needs to be removed from page + */ + data_deleted_last+= move_length; + } + + /* First log changes to page */ + log_pos= log_data + FILEID_STORE_SIZE; + page_store(log_pos, page); + log_pos+= PAGE_STORE_SIZE; + +#ifdef EXTRA_DEBUG_KEY_CHANGES + *log_pos++= KEY_OP_DEBUG; + *log_pos++= KEY_OP_DEBUG_LOG_MIDDLE; +#endif + + /* Store keypage_flag */ + *log_pos++= KEY_OP_SET_PAGEFLAG; + *log_pos++= _ma_get_keypage_flag(info->s, ma_page->buff); + + log_pos[0]= KEY_OP_DEL_SUFFIX; + int2store(log_pos+1, data_deleted_last); + log_pos+= 3; + + log_pos[0]= KEY_OP_ADD_PREFIX; + int2store(log_pos+1, data_added_first); + int2store(log_pos+3, data_changed_first); + log_pos+= 5; + + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - + log_data); + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= (ma_page->buff + + info->s->keypage_header); + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= data_changed_first; + translog_parts= 2; + extra_length= data_changed_first; + + /* If changed key is on page, log those changes too */ + + if (key_offset < new_length) + { + uchar *start_log_pos= log_pos; + + log_pos[0]= KEY_OP_OFFSET; + int2store(log_pos+1, key_offset); + log_pos+= 3; + if (move_length) + { + log_pos[0]= KEY_OP_SHIFT; + int2store(log_pos+1, move_length); + log_pos+= 3; + } + log_pos[0]= KEY_OP_CHANGE; + int2store(log_pos+1, key_length); + log_pos+= 3; + + log_array[TRANSLOG_INTERNAL_PARTS + 2].str= start_log_pos; + log_array[TRANSLOG_INTERNAL_PARTS + 2].length= (uint) (log_pos - + start_log_pos); + + log_array[TRANSLOG_INTERNAL_PARTS + 3].str= key_pos; + log_array[TRANSLOG_INTERNAL_PARTS + 3].length= key_length; + translog_parts+=2; + extra_length+= (uint) (log_array[TRANSLOG_INTERNAL_PARTS + 2].length + + key_length); + } + + _ma_log_key_changes(ma_page, + log_array + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_pos, &extra_length, &translog_parts); + /* Remember new page length for future log entires for same page */ + ma_page->org_size= ma_page->size; + + DBUG_RETURN(translog_write_record(&lsn, LOGREC_REDO_INDEX, + info->trn, info, + (translog_size_t) + (log_array[TRANSLOG_INTERNAL_PARTS + + 0].length + extra_length), + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_array, log_data, NULL)); +} + + +#ifdef NOT_NEEDED + +/** + @brief + Write log entry for page that has got data added first and + data deleted last +*/ + +static my_bool _ma_log_middle(MARIA_PAGE *ma_page, + uint data_added_first, uint data_changed_first, + uint data_deleted_last) +{ + LSN lsn; + LEX_STRING log_array[TRANSLOG_INTERNAL_PARTS + 4]; + uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 3 + 5 + 7], *log_pos; + MARIA_HA *info= ma_page->info; + my_off_t page= ma_page->page / info->s->block_size; + uint translog_parts, extra_length; + DBUG_ENTER("_ma_log_middle"); + DBUG_PRINT("enter", ("page: %lu", (ulong) page)); + + DBUG_ASSERT(ma_page->org_size + data_added_first - data_deleted_last == + ma_page->size); + + log_pos= log_data + FILEID_STORE_SIZE; + page_store(log_pos, page); + log_pos+= PAGE_STORE_SIZE; + + log_pos[0]= KEY_OP_DEL_PREFIX; + int2store(log_pos+1, data_deleted_last); + log_pos+= 3; + + log_pos[0]= KEY_OP_ADD_PREFIX; + int2store(log_pos+1, data_added_first); + int2store(log_pos+3, data_changed_first); + log_pos+= 5; + + log_array[TRANSLOG_INTERNAL_PARTS + 0].str= log_data; + log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - + log_data); + + log_array[TRANSLOG_INTERNAL_PARTS + 1].str= ((char*) buff + + info->s->keypage_header); + log_array[TRANSLOG_INTERNAL_PARTS + 1].length= data_changed_first; + translog_parts= 2; + extra_length= data_changed_first; + + _ma_log_key_changes(ma_page, + log_array + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_pos, &extra_length, &translog_parts); + /* Remember new page length for future log entires for same page */ + ma_page->org_size= ma_page->size; + + DBUG_RETURN(translog_write_record(&lsn, LOGREC_REDO_INDEX, + info->trn, info, + (translog_size_t) + log_array[TRANSLOG_INTERNAL_PARTS + + 0].length + extra_length, + TRANSLOG_INTERNAL_PARTS + translog_parts, + log_array, log_data, NULL)); +} +#endif |