Adding upstream version 1:10.11.6.upstream/1%10.11.6upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1699 insertions, 0 deletions

diff --git a/storage/maria/ma_delete.c b/storage/maria/ma_delete.c
new file mode 100644
index 00000000..b24cfcc6
--- /dev/null
+++ b/storage/maria/ma_delete.c
@@ -0,0 +1,1699 @@
+/* Copyright (C) 2006 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
+   Copyright (C) 2009-2010 Monty Program 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 */
+
+#include "ma_fulltext.h"
+#include "ma_rt_index.h"
+#include "trnman.h"
+#include "ma_key_recover.h"
+
+static int d_search(MARIA_HA *info, MARIA_KEY *key, uint32 comp_flag,
+                    MARIA_PAGE *page);
+static int del(MARIA_HA *info, MARIA_KEY *key,
+               MARIA_PAGE *anc_page, MARIA_PAGE *leaf_page,
+	       uchar *keypos, my_off_t next_block, uchar *ret_key_buff);
+static int underflow(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
+		     MARIA_PAGE *anc_page, MARIA_PAGE *leaf_page,
+		     uchar *keypos);
+static uint remove_key(MARIA_KEYDEF *keyinfo, uint page_flag, uint nod_flag,
+                       uchar *keypos, uchar *lastkey, uchar *page_end,
+		       my_off_t *next_block, MARIA_KEY_PARAM *s_temp);
+
+/* @breif Remove a row from a MARIA table */
+
+int maria_delete(MARIA_HA *info,const uchar *record)
+{
+  uint i;
+  uchar *old_key;
+  int save_errno;
+  char lastpos[8];
+  MARIA_SHARE *share= info->s;
+  MARIA_KEYDEF *keyinfo;
+  DBUG_ENTER("maria_delete");
+
+  /* Test if record is in datafile */
+  DBUG_EXECUTE_IF("maria_pretend_crashed_table_on_usage",
+                  _ma_print_error(info, HA_ERR_CRASHED, 0);
+                  DBUG_RETURN(my_errno= HA_ERR_CRASHED););
+  DBUG_EXECUTE_IF("my_error_test_undefined_error",
+                  _ma_print_error(info, INT_MAX, 0);
+                  DBUG_RETURN(my_errno= INT_MAX););
+  if (!(info->update & HA_STATE_AKTIV))
+  {
+    DBUG_RETURN(my_errno=HA_ERR_KEY_NOT_FOUND);	/* No database read */
+  }
+  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->compare_record)(info,record))
+    goto err;				/* Error on read-check */
+
+  if (_ma_mark_file_changed(share))
+    goto err;
+
+  /* Ensure we don't change the autoincrement value */
+  info->last_auto_increment= ~(ulonglong) 0;
+  /* Remove all keys from the index file */
+
+  old_key= info->lastkey_buff2;
+
+  for (i=0, keyinfo= share->keyinfo ; i < share->base.keys ; i++, keyinfo++)
+  {
+    if (maria_is_key_active(share->state.key_map, i))
+    {
+      keyinfo->version++;
+      if (keyinfo->flag & HA_FULLTEXT)
+      {
+        if (_ma_ft_del(info, i, old_key, record, info->cur_row.lastpos))
+          goto err;
+      }
+      else
+      {
+        MARIA_KEY key;
+        if (keyinfo->ck_delete(info,
+                               (*keyinfo->make_key)(info, &key, i, old_key,
+                                                    record,
+                                                    info->cur_row.lastpos,
+                                                    info->cur_row.trid)))
+          goto err;
+      }
+      /* The above changed info->lastkey2. Inform maria_rnext_same(). */
+      info->update&= ~HA_STATE_RNEXT_SAME;
+    }
+  }
+
+  if (share->calc_checksum)
+  {
+    /*
+      We can't use the row based checksum as this doesn't have enough
+      precision.
+    */
+    info->cur_row.checksum= (*share->calc_checksum)(info, record);
+  }
+
+  if ((*share->delete_record)(info, record))
+    goto err;				/* Remove record from database */
+
+  info->state->checksum-= info->cur_row.checksum;
+  info->state->records--;
+  info->update= HA_STATE_CHANGED+HA_STATE_DELETED+HA_STATE_ROW_CHANGED;
+  info->row_changes++;
+  share->state.changed|= (STATE_NOT_OPTIMIZED_ROWS | STATE_NOT_MOVABLE |
+                          STATE_NOT_ZEROFILLED);
+  info->state->changed=1;
+
+  mi_sizestore(lastpos, info->cur_row.lastpos);
+  _ma_writeinfo(info, WRITEINFO_UPDATE_KEYFILE);
+  if (info->invalidator != 0)
+  {
+    DBUG_PRINT("info", ("invalidator... '%s' (delete)",
+                        share->open_file_name.str));
+    (*info->invalidator)(share->open_file_name.str);
+    info->invalidator=0;
+  }
+  DBUG_RETURN(0);
+
+err:
+  save_errno= my_errno;
+  DBUG_ASSERT(save_errno);
+  if (!save_errno)
+    save_errno= HA_ERR_INTERNAL_ERROR;          /* Should never happen */
+
+  mi_sizestore(lastpos, info->cur_row.lastpos);
+  (void) _ma_writeinfo(info,WRITEINFO_UPDATE_KEYFILE);
+  info->update|=HA_STATE_WRITTEN;	/* Buffer changed */
+  if (save_errno != HA_ERR_RECORD_CHANGED)
+  {
+    _ma_set_fatal_error(info, HA_ERR_CRASHED);
+    save_errno= HA_ERR_CRASHED;
+  }
+  DBUG_RETURN(my_errno= save_errno);
+} /* maria_delete */
+
+
+/*
+  Remove a key from the btree index
+
+  TODO:
+   Change ma_ck_real_delete() to use another buffer for changed keys instead
+   of key->data. This would allows us to remove the copying of the key here.
+*/
+
+my_bool _ma_ck_delete(MARIA_HA *info, MARIA_KEY *key)
+{
+  MARIA_SHARE *share= info->s;
+  int res;
+  my_bool buff_alloced;
+  LSN lsn= LSN_IMPOSSIBLE;
+  my_off_t new_root= share->state.key_root[key->keyinfo->key_nr];
+  uchar *key_buff, *save_key_data;
+  MARIA_KEY org_key;
+  DBUG_ENTER("_ma_ck_delete");
+
+  LINT_INIT_STRUCT(org_key);
+
+  alloc_on_stack(*info->stack_end_ptr, key_buff, buff_alloced,
+                 key->keyinfo->max_store_length);
+  if (!key_buff)
+    DBUG_RETURN(1);
+
+  save_key_data= key->data;
+  if (share->now_transactional)
+  {
+    /* Save original value as the key may change */
+    memcpy(key_buff, key->data, key->data_length + key->ref_length);
+    org_key= *key;
+    key->data= key_buff;
+  }
+
+  if ((res= _ma_ck_real_delete(info, key, &new_root)))
+  {
+    /* We have to mark the table crashed before unpin_all_pages() */
+    maria_mark_crashed(info);
+  }
+
+  key->data= save_key_data;
+  if (!res && share->now_transactional)
+    res= _ma_write_undo_key_delete(info, &org_key, new_root, &lsn);
+  else
+  {
+    share->state.key_root[key->keyinfo->key_nr]= new_root;
+    _ma_fast_unlock_key_del(info);
+  }
+  _ma_unpin_all_pages_and_finalize_row(info, lsn);
+
+  stack_alloc_free(key_buff, buff_alloced);
+  DBUG_RETURN(res != 0);
+} /* _ma_ck_delete */
+
+
+my_bool _ma_ck_real_delete(register MARIA_HA *info, MARIA_KEY *key,
+                           my_off_t *root)
+{
+  int error;
+  my_bool result= 0, buff_alloced;
+  my_off_t old_root;
+  uchar *root_buff;
+  MARIA_KEYDEF *keyinfo= key->keyinfo;
+  MARIA_PAGE page;
+  DBUG_ENTER("_ma_ck_real_delete");
+
+  if ((old_root=*root) == HA_OFFSET_ERROR)
+  {
+    _ma_set_fatal_error(info, HA_ERR_CRASHED);
+    DBUG_RETURN(1);
+  }
+
+  alloc_on_stack(*info->stack_end_ptr, root_buff, buff_alloced,
+                 (keyinfo->block_length + keyinfo->max_store_length*2));
+  if (!root_buff)
+    DBUG_RETURN(1);
+
+  DBUG_PRINT("info",("root_page: %lu",
+                     (ulong) (old_root / keyinfo->block_length)));
+  if (_ma_fetch_keypage(&page, info, keyinfo, old_root,
+                        PAGECACHE_LOCK_WRITE, DFLT_INIT_HITS, root_buff, 0))
+  {
+    result= 1;
+    goto err;
+  }
+  if ((error= d_search(info, key, (keyinfo->flag & HA_FULLTEXT ?
+                                   SEARCH_FIND | SEARCH_UPDATE | SEARCH_INSERT:
+                                   SEARCH_SAME),
+                       &page)))
+  {
+    if (error < 0)
+      result= 1;
+    else if (error == 2)
+    {
+      DBUG_PRINT("test",("Enlarging of root when deleting"));
+      if (_ma_enlarge_root(info, key, root))
+        result= 1;
+    }
+    else /* error == 1 */
+    {
+      MARIA_SHARE *share= info->s;
+
+      page_mark_changed(info, &page);
+
+      if (page.size <= page.node + share->keypage_header + 1)
+      {
+        DBUG_ASSERT(page.size == page.node + share->keypage_header);
+	if (page.node)
+	  *root= _ma_kpos(page.node, root_buff +share->keypage_header +
+                          page.node);
+	else
+	  *root=HA_OFFSET_ERROR;
+	if (_ma_dispose(info, old_root, 0))
+	  result= 1;
+      }
+      else if (_ma_write_keypage(&page, PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                                 DFLT_INIT_HITS))
+        result= 1;
+    }
+  }
+err:
+  stack_alloc_free(root_buff, buff_alloced);
+  DBUG_PRINT("exit",("Return: %d",result));
+  DBUG_RETURN(result);
+} /* _ma_ck_real_delete */
+
+
+/**
+   @brief Remove key below key root
+
+   @param key  Key to delete.  Will contain new key if block was enlarged
+
+   @return
+   @retval 0   ok (anc_page is not changed)
+   @retval 1   If data on page is too small; In this case anc_buff is not saved
+   @retval 2   If data on page is too big
+   @retval -1  On errors
+*/
+
+static int d_search(MARIA_HA *info, MARIA_KEY *key, uint32 comp_flag,
+                    MARIA_PAGE *anc_page)
+{
+  int flag,ret_value,save_flag;
+  uint nod_flag, page_flag;
+  my_bool last_key, buff_alloced= 0, lastkey_alloced;
+  uchar *leaf_buff=0, *keypos, *lastkey;
+  MARIA_KEY_PARAM s_temp;
+  MARIA_SHARE *share= info->s;
+  MARIA_KEYDEF *keyinfo= key->keyinfo;
+  MARIA_PAGE leaf_page;
+  DBUG_ENTER("d_search");
+  DBUG_DUMP("page", anc_page->buff, anc_page->size);
+
+  alloc_on_stack(*info->stack_end_ptr, lastkey, lastkey_alloced,
+                 keyinfo->max_store_length);
+  if (!lastkey)
+    DBUG_RETURN(1);
+
+  flag=(*keyinfo->bin_search)(key, anc_page, comp_flag, &keypos, lastkey,
+                              &last_key);
+  if (flag == MARIA_FOUND_WRONG_KEY)
+  {
+    DBUG_PRINT("error",("Found wrong key"));
+    goto err;
+  }
+  page_flag= anc_page->flag;
+  nod_flag=  anc_page->node;
+
+  if (!flag && (keyinfo->flag & HA_FULLTEXT))
+  {
+    uint off;
+    int  subkeys;
+
+    get_key_full_length_rdonly(off, lastkey);
+    subkeys=ft_sintXkorr(lastkey+off);
+    DBUG_ASSERT(info->ft1_to_ft2==0 || subkeys >=0);
+    comp_flag=SEARCH_SAME;
+    if (subkeys >= 0)
+    {
+      /* normal word, one-level tree structure */
+      if (info->ft1_to_ft2)
+      {
+        /* we're in ft1->ft2 conversion mode. Saving key data */
+        insert_dynamic(info->ft1_to_ft2, (lastkey+off));
+      }
+      else
+      {
+        /* we need exact match only if not in ft1->ft2 conversion mode */
+        flag=(*keyinfo->bin_search)(key, anc_page, comp_flag, &keypos,
+                                    lastkey, &last_key);
+      }
+      /* fall through to normal delete */
+    }
+    else
+    {
+      /* popular word. two-level tree. going down */
+      uint tmp_key_length;
+      my_off_t root;
+      uchar *kpos=keypos;
+      MARIA_KEY tmp_key;
+
+      tmp_key.data=    lastkey;
+      tmp_key.keyinfo= keyinfo;
+
+      if (!(tmp_key_length=(*keyinfo->get_key)(&tmp_key, page_flag, nod_flag,
+                                               &kpos)))
+      {
+        _ma_set_fatal_error(info, HA_ERR_CRASHED);
+        goto err;
+      }
+      root= _ma_row_pos_from_key(&tmp_key);
+      if (subkeys == -1)
+      {
+        /* the last entry in sub-tree */
+        if (_ma_dispose(info, root, 1))
+          goto err;
+        /* fall through to normal delete */
+      }
+      else
+      {
+        MARIA_KEY word_key;
+        keyinfo=&share->ft2_keyinfo;
+        /* we'll modify key entry 'in vivo' */
+        kpos-=keyinfo->keylength+nod_flag;
+        get_key_full_length_rdonly(off, key->data);
+
+        word_key.data=        key->data + off;
+        word_key.keyinfo=     &share->ft2_keyinfo;
+        word_key.data_length= HA_FT_WLEN;
+        word_key.ref_length= 0;
+        word_key.flag= 0;
+        ret_value= _ma_ck_real_delete(info, &word_key, &root);
+        _ma_dpointer(share, kpos+HA_FT_WLEN, root);
+        subkeys++;
+        ft_intXstore(kpos, subkeys);
+        if (!ret_value)
+        {
+          page_mark_changed(info, anc_page);
+          ret_value= _ma_write_keypage(anc_page,
+                                       PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                                       DFLT_INIT_HITS);
+        }
+        goto end;
+      }
+    }
+  }
+  if (nod_flag)
+  {
+    /* Read left child page */
+    leaf_page.pos= _ma_kpos(nod_flag,keypos);
+
+    alloc_on_stack(*info->stack_end_ptr, leaf_buff, buff_alloced,
+                   (keyinfo->block_length + keyinfo->max_store_length*2));
+    if (!leaf_buff)
+      goto err;
+
+    if (_ma_fetch_keypage(&leaf_page, info,keyinfo, leaf_page.pos,
+                          PAGECACHE_LOCK_WRITE, DFLT_INIT_HITS, leaf_buff,
+                          0))
+      goto err;
+  }
+
+  if (flag != 0)
+  {
+    if (!nod_flag)
+    {
+      /* This should newer happend */
+      DBUG_PRINT("error",("Didn't find key"));
+      _ma_set_fatal_error(info, HA_ERR_CRASHED);
+      goto err;
+    }
+    save_flag=0;
+    ret_value= d_search(info, key, comp_flag, &leaf_page);
+  }
+  else
+  {						/* Found key */
+    uint tmp;
+    uint anc_buff_length= anc_page->size;
+    uint anc_page_flag=   anc_page->flag;
+    my_off_t next_block;
+
+    if (!(tmp= remove_key(keyinfo, anc_page_flag, nod_flag, keypos, lastkey,
+                          anc_page->buff + anc_buff_length,
+                          &next_block, &s_temp)))
+      goto err;
+
+    page_mark_changed(info, anc_page);
+    anc_buff_length-= tmp;
+    anc_page->size= anc_buff_length;
+    page_store_size(share, anc_page);
+
+    /*
+      Log initial changes on pages
+      If there is an underflow, there will be more changes logged to the
+      page
+    */
+    if (share->now_transactional &&
+        _ma_log_delete(anc_page, s_temp.key_pos,
+                       s_temp.changed_length, s_temp.move_length,
+                       0, KEY_OP_DEBUG_LOG_DEL_CHANGE_1))
+      goto err;
+
+    if (!nod_flag)
+    {						/* On leaf page */
+      if (anc_buff_length <= (info->quick_mode ?
+                              MARIA_MIN_KEYBLOCK_LENGTH :
+                              (uint) keyinfo->underflow_block_length))
+      {
+        /* Page will be written by caller if we return 1 */
+        ret_value= 1;
+        goto end;
+      }
+      if (_ma_write_keypage(anc_page,
+                            PAGECACHE_LOCK_LEFT_WRITELOCKED, DFLT_INIT_HITS))
+	goto err;
+
+      ret_value= 0;                             /* Return ok */
+      goto end;
+    }
+    save_flag=1;                         /* Mark that anc_buff is changed */
+    ret_value= del(info, key, anc_page, &leaf_page,
+                   keypos, next_block, lastkey);
+  }
+  if (ret_value >0)
+  {
+    save_flag= 2;
+    if (ret_value == 1)
+      ret_value= underflow(info, keyinfo, anc_page, &leaf_page, keypos);
+    else
+    {
+      /* This can only happen with variable length keys */
+      MARIA_KEY last_key;
+      DBUG_PRINT("test",("Enlarging of key when deleting"));
+
+      last_key.data=    lastkey;
+      last_key.keyinfo= keyinfo;
+      if (!_ma_get_last_key(&last_key, anc_page, keypos))
+	goto err;
+      ret_value= _ma_insert(info, key, anc_page, keypos,
+                            last_key.data,
+                            (MARIA_PAGE*) 0, (uchar*) 0, (my_bool) 0);
+
+      if (_ma_write_keypage(&leaf_page, PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                            DFLT_INIT_HITS))
+        ret_value= -1;
+    }
+  }
+  if (ret_value == 0 && anc_page->size > share->max_index_block_size)
+  {
+    /*
+      parent buffer got too big ; We have to split the page.
+      The | 2 is there to force write of anc page below
+    */
+    save_flag= 3;
+    ret_value= _ma_split_page(info, key, anc_page,
+                              share->max_index_block_size,
+                              (uchar*) 0, 0, 0, lastkey, 0) | 2;
+    DBUG_ASSERT(anc_page->org_size == anc_page->size);
+  }
+  if (save_flag && ret_value != 1)
+  {
+    page_mark_changed(info, anc_page);
+    if (_ma_write_keypage(anc_page, PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                          DFLT_INIT_HITS))
+      ret_value= -1;
+  }
+  else
+  {
+    DBUG_DUMP("page", anc_page->buff, anc_page->size);
+  }
+
+end:
+  stack_alloc_free(leaf_buff, buff_alloced);
+  stack_alloc_free(lastkey, lastkey_alloced);
+  DBUG_PRINT("exit",("Return: %d",ret_value));
+  DBUG_RETURN(ret_value);
+
+err:
+  stack_alloc_free(leaf_buff, buff_alloced);
+  stack_alloc_free(lastkey, lastkey_alloced);
+  DBUG_PRINT("exit",("Error: %d",my_errno));
+  DBUG_RETURN (-1);
+} /* d_search */
+
+
+/**
+   @brief Remove a key that has a page-reference
+
+   @param info		 Maria handler
+   @param key		 Buffer for key to be inserted at upper level
+   @param anc_page	 Page address for page where deleted key was
+   @param anc_buff       Page buffer (nod) where deleted key was
+   @param leaf_page      Page address for nod before the deleted key
+   @param leaf_buff      Buffer for leaf_page
+   @param leaf_buff_link Pinned page link for leaf_buff
+   @param keypos         Pos to where deleted key was on anc_buff
+   @param next_block	 Page adress for nod after deleted key
+   @param ret_key_buff	 Key before keypos in anc_buff
+
+   @notes
+      leaf_page must be written to disk if retval > 0
+      anc_page  is not updated on disk. Caller should do this
+
+   @return
+   @retval < 0   Error
+   @retval 0     OK.    leaf_buff is written to disk
+
+   @retval 1     key contains key to upper level (from balance page)
+                 leaf_buff has underflow
+   @retval 2     key contains key to upper level (from split space)
+*/
+
+static int del(MARIA_HA *info, MARIA_KEY *key,
+               MARIA_PAGE *anc_page, MARIA_PAGE *leaf_page,
+	       uchar *keypos, my_off_t next_block, uchar *ret_key_buff)
+{
+  int ret_value,length;
+  uint a_length, page_flag, nod_flag, leaf_length, new_leaf_length;
+  uchar *keybuff,*endpos,*next_buff,*key_start, *prev_key;
+  uchar *anc_buff;
+  my_bool buff_alloced= 0, keybuff_alloced;
+  MARIA_KEY_PARAM s_temp;
+  MARIA_KEY tmp_key;
+  MARIA_SHARE *share= info->s;
+  MARIA_KEYDEF *keyinfo= key->keyinfo;
+  MARIA_KEY ret_key;
+  MARIA_PAGE next_page;
+  DBUG_ENTER("del");
+  DBUG_PRINT("enter",("leaf_page: %lu  keypos: %p",
+                      (ulong) (leaf_page->pos / share->block_size),
+		      keypos));
+  DBUG_DUMP("leaf_buff", leaf_page->buff, leaf_page->size);
+
+  alloc_on_stack(*info->stack_end_ptr, keybuff, keybuff_alloced,
+                 keyinfo->max_store_length);
+  if (!keybuff)
+    DBUG_RETURN(1);
+
+  page_flag=   leaf_page->flag;
+  leaf_length= leaf_page->size;
+  nod_flag=    leaf_page->node;
+
+  endpos= leaf_page->buff + leaf_length;
+  tmp_key.keyinfo= keyinfo;
+  tmp_key.data=    keybuff;
+  next_buff= 0;
+
+  if (!(key_start= _ma_get_last_key(&tmp_key, leaf_page, endpos)))
+    goto err;
+
+  if (nod_flag)
+  {
+    next_page.pos= _ma_kpos(nod_flag,endpos);
+
+    alloc_on_stack(*info->stack_end_ptr, next_buff, buff_alloced,
+                   (keyinfo->block_length + keyinfo->max_store_length*2));
+    if (!next_buff)
+      goto err;
+
+    if (_ma_fetch_keypage(&next_page, info, keyinfo, next_page.pos,
+                          PAGECACHE_LOCK_WRITE, DFLT_INIT_HITS, next_buff, 0))
+      ret_value= -1;
+    else
+    {
+      DBUG_DUMP("next_page", next_page.buff, next_page.size);
+      if ((ret_value= del(info, key, anc_page, &next_page,
+                          keypos, next_block, ret_key_buff)) >0)
+      {
+        /* Get new length after key was deleted */
+	endpos= leaf_page->buff+ leaf_page->size;
+	if (ret_value == 1)
+	{
+          /* underflow writes "next_page" to disk */
+	  ret_value= underflow(info, keyinfo, leaf_page, &next_page,
+                               endpos);
+          if (ret_value < 0)
+            goto err;
+	  if (leaf_page->size > share->max_index_block_size)
+	  {
+            DBUG_ASSERT(ret_value == 0);
+	    ret_value= (_ma_split_page(info, key, leaf_page,
+                                       share->max_index_block_size,
+                                       (uchar*) 0, 0, 0,
+                                       ret_key_buff, 0) | 2);
+	  }
+	}
+	else
+	{
+          if (_ma_write_keypage(&next_page, PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                                DFLT_INIT_HITS))
+            goto err;
+	  DBUG_PRINT("test",("Inserting of key when deleting"));
+	  if (!_ma_get_last_key(&tmp_key, leaf_page, endpos))
+	    goto err;
+	  ret_value= _ma_insert(info, key, leaf_page, endpos,
+                                tmp_key.data, (MARIA_PAGE *) 0, (uchar*) 0,
+                                0);
+	}
+      }
+      page_mark_changed(info, leaf_page);
+      /*
+        If ret_value <> 0, then leaf_page underflowed and caller will have
+        to handle underflow and write leaf_page to disk.
+        We can't write it here, as if leaf_page is empty we get an assert
+        in _ma_write_keypage.
+      */
+      if (ret_value == 0 && _ma_write_keypage(leaf_page,
+                                              PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                                              DFLT_INIT_HITS))
+	goto err;
+    }
+    stack_alloc_free(next_buff, buff_alloced);
+    stack_alloc_free(keybuff, keybuff_alloced);
+    DBUG_ASSERT(leaf_page->size <= share->max_index_block_size);
+    DBUG_RETURN(ret_value);
+  }
+
+  /*
+    Remove last key from leaf page
+    Note that leaf_page page may only have had one key (can normally only
+    happen in quick mode), in which ase it will now temporary have 0 keys
+    on it. This will be corrected by the caller as we will return 0.
+  */
+  new_leaf_length= (uint) (key_start - leaf_page->buff);
+  leaf_page->size= new_leaf_length;
+  page_store_size(share, leaf_page);
+
+  if (share->now_transactional &&
+      _ma_log_suffix(leaf_page, leaf_length, new_leaf_length))
+    goto err;
+
+  page_mark_changed(info, leaf_page);           /* Safety */
+  if (new_leaf_length <= (info->quick_mode ? MARIA_MIN_KEYBLOCK_LENGTH :
+                          (uint) keyinfo->underflow_block_length))
+  {
+    /* Underflow, leaf_page will be written by caller */
+    ret_value= 1;
+  }
+  else
+  {
+    ret_value= 0;
+    if (_ma_write_keypage(leaf_page, PAGECACHE_LOCK_LEFT_WRITELOCKED,
+                          DFLT_INIT_HITS))
+      goto err;
+  }
+
+  /* Place last key in ancestor page on deleted key position */
+  a_length= anc_page->size;
+  anc_buff= anc_page->buff;
+  endpos=   anc_buff + a_length;
+
+  ret_key.keyinfo= keyinfo;
+  ret_key.data=    ret_key_buff;
+
+  prev_key= 0;
+  if (keypos != anc_buff+share->keypage_header + share->base.key_reflength)
+  {
+    if (!_ma_get_last_key(&ret_key, anc_page, keypos))
+      goto err;
+    prev_key= ret_key.data;
+  }
+  length= (*keyinfo->pack_key)(&tmp_key, share->base.key_reflength,
+                               keypos == endpos ? (uchar*) 0 : keypos,
+                               prev_key, prev_key,
+                               &s_temp);
+  if (length > 0)
+    bmove_upp(endpos+length,endpos,(uint) (endpos-keypos));
+  else
+    bmove(keypos,keypos-length, (int) (endpos-keypos)+length);
+  (*keyinfo->store_key)(keyinfo,keypos,&s_temp);
+  key_start= keypos;
+  if (tmp_key.flag & (SEARCH_USER_KEY_HAS_TRANSID |
+                      SEARCH_PAGE_KEY_HAS_TRANSID))
+    _ma_mark_page_with_transid(share, anc_page);
+
+  /* Save pointer to next leaf on parent page */
+  if (!(*keyinfo->get_key)(&ret_key, page_flag, share->base.key_reflength,
+                           &keypos))
+    goto err;
+  _ma_kpointer(info,keypos - share->base.key_reflength,next_block);
+  anc_page->size= a_length + length;
+  page_store_size(share, anc_page);
+
+  if (share->now_transactional &&
+      _ma_log_add(anc_page, a_length,
+                  key_start, s_temp.changed_length, s_temp.move_length, 1,
+                  KEY_OP_DEBUG_LOG_ADD_2))
+    goto err;
+
+  DBUG_ASSERT(leaf_page->size <= share->max_index_block_size);
+  stack_alloc_free(next_buff, buff_alloced);
+  stack_alloc_free(keybuff, keybuff_alloced);
+  DBUG_RETURN(new_leaf_length <=
+              (info->quick_mode ? MARIA_MIN_KEYBLOCK_LENGTH :
+               (uint) keyinfo->underflow_block_length));
+
+err:
+  stack_alloc_free(next_buff, buff_alloced);
+  stack_alloc_free(keybuff, keybuff_alloced);
+  DBUG_RETURN(-1);
+} /* del */
+
+
+/**
+   @brief Balances adjacent pages if underflow occours
+
+   @fn    underflow()
+   @param anc_buff        Anchestor page data
+   @param leaf_page       Leaf page (page that underflowed)
+   @param leaf_page_link  Pointer to pin information about leaf page
+   @param keypos          Position after current key in anc_buff
+
+   @note
+     This function writes redo entries for all changes
+     leaf_page is saved to disk
+     Caller must save anc_buff
+
+     For the algoritm to work, we have to ensure for packed keys that
+     key_length + (underflow_length + max_block_length + key_length) / 2
+     <= block_length.
+     From which follows that underflow_length <= block_length - key_length *3
+     For not packed keys we have:
+     (underflow_length + max_block_length + key_length) / 2 <= block_length
+     From which follows that underflow_length < block_length - key_length
+     This is ensured by setting of underflow_block_length.
+
+   @return
+   @retval  0  ok
+   @retval  1  ok, but anc_page did underflow
+   @retval -1  error
+ */
+
+static int underflow(MARIA_HA *info, MARIA_KEYDEF *keyinfo,
+		     MARIA_PAGE *anc_page, MARIA_PAGE *leaf_page,
+		     uchar *keypos)
+{
+  int t_length;
+  uint anc_length,buff_length,leaf_length,p_length,s_length,nod_flag;
+  uint next_buff_length, new_buff_length, key_reflength;
+  uint unchanged_leaf_length, new_leaf_length, new_anc_length;
+  uint anc_page_flag, page_flag;
+  uchar *anc_key_buff, *leaf_key_buff;
+  uchar *endpos, *next_keypos, *anc_pos, *half_pos, *prev_key;
+  uchar *anc_buff, *leaf_buff;
+  uchar *after_key, *anc_end_pos;
+  MARIA_KEY_PARAM key_deleted, key_inserted;
+  MARIA_SHARE *share= info->s;
+  my_bool first_key, buff_alloced;
+  MARIA_KEY tmp_key, anc_key, leaf_key;
+  MARIA_PAGE next_page;
+  DBUG_ENTER("underflow");
+  DBUG_PRINT("enter",("leaf_page: %lu  keypos: %p",
+                      (ulong) (leaf_page->pos / share->block_size),
+		      keypos));
+  DBUG_DUMP("anc_buff", anc_page->buff,  anc_page->size);
+  DBUG_DUMP("leaf_buff", leaf_page->buff, leaf_page->size);
+
+  alloc_on_stack(*info->stack_end_ptr, anc_key_buff, buff_alloced,
+                 keyinfo->max_store_length*2);
+  if (!anc_key_buff)
+    DBUG_RETURN(1);
+
+  leaf_key_buff= anc_key_buff+ keyinfo->max_store_length;
+
+  anc_page_flag= anc_page->flag;
+  anc_buff= anc_page->buff;
+  leaf_buff= leaf_page->buff;
+  info->keyread_buff_used=1;
+  next_keypos=keypos;
+  nod_flag= leaf_page->node;
+  p_length= nod_flag+share->keypage_header;
+  anc_length= anc_page->size;
+  leaf_length= leaf_page->size;
+  key_reflength= share->base.key_reflength;
+  if (share->keyinfo+info->lastinx == keyinfo)
+    info->page_changed=1;
+  first_key= keypos == anc_buff + share->keypage_header + key_reflength;
+
+  tmp_key.data=  info->buff;
+  anc_key.data=  anc_key_buff;
+  leaf_key.data= leaf_key_buff;
+  tmp_key.keyinfo= leaf_key.keyinfo= anc_key.keyinfo= keyinfo;
+
+  if ((keypos < anc_buff + anc_length && (info->state->records & 1)) ||
+      first_key)
+  {
+    uint tmp_length;
+    uint next_page_flag;
+    /* Use page right of anc-page */
+    DBUG_PRINT("test",("use right page"));
+
+    /*
+      Calculate position after the current key. Note that keydata itself is
+      not used
+    */
+    if (keyinfo->flag & HA_BINARY_PACK_KEY)
+    {
+      if (!(next_keypos= _ma_get_key(&tmp_key, anc_page, keypos)))
+	goto err;
+    }
+    else
+    {
+      /* Avoid length error check if packed key */
+      tmp_key.data[0]= tmp_key.data[1]= 0;
+      /* Got to end of found key */
+      if (!(*keyinfo->get_key)(&tmp_key, anc_page_flag, key_reflength,
+                               &next_keypos))
+        goto err;
+    }
+    next_page.pos= _ma_kpos(key_reflength, next_keypos);
+    if (_ma_fetch_keypage(&next_page, info, keyinfo, next_page.pos,
+                          PAGECACHE_LOCK_WRITE, DFLT_INIT_HITS, info->buff, 0))
+      goto err;
+    next_buff_length= next_page.size;
+    next_page_flag=   next_page.flag;
+    DBUG_DUMP("next", next_page.buff, next_page.size);
+
+    /* find keys to make a big key-page */
+    bmove(next_keypos-key_reflength, next_page.buff + share->keypage_header,
+          key_reflength);
+
+    if (!_ma_get_last_key(&anc_key, anc_page, next_keypos) ||
+	!_ma_get_last_key(&leaf_key, leaf_page, leaf_buff+leaf_length))
+      goto err;
+
+    /* merge pages and put parting key from anc_page between */
+    prev_key= (leaf_length == p_length ? (uchar*) 0 : leaf_key.data);
+    t_length= (*keyinfo->pack_key)(&anc_key, nod_flag, next_page.buff+p_length,
+                                   prev_key, prev_key, &key_inserted);
+    tmp_length= next_buff_length - p_length;
+    endpos= next_page.buff + tmp_length + leaf_length + t_length;
+    /* next_page.buff will always be larger than before !*/
+    bmove_upp(endpos, next_page.buff + next_buff_length, tmp_length);
+    memcpy(next_page.buff, leaf_buff,(size_t) leaf_length);
+    (*keyinfo->store_key)(keyinfo, next_page.buff+leaf_length, &key_inserted);
+    buff_length= (uint) (endpos - next_page.buff);
+
+    /* Set page flag from combination of both key pages and parting key */
+    page_flag= next_page_flag | leaf_page->flag;
+    if (anc_key.flag & (SEARCH_USER_KEY_HAS_TRANSID |
+                        SEARCH_PAGE_KEY_HAS_TRANSID))
+      page_flag|= KEYPAGE_FLAG_HAS_TRANSID;
+
+    next_page.size= buff_length;
+    next_page.flag= page_flag;
+    page_store_info(share, &next_page);
+
+    /* remove key from anc_page */
+    if (!(s_length=remove_key(keyinfo, anc_page_flag, key_reflength, keypos,
+                              anc_key_buff, anc_buff+anc_length,
+                              (my_off_t *) 0, &key_deleted)))
+      goto err;
+
+    new_anc_length= anc_length - s_length;
+    anc_page->size= new_anc_length;
+    page_store_size(share, anc_page);
+
+    if (buff_length <= share->max_index_block_size)
+    {
+      /* All keys fitted into one page */
+      page_mark_changed(info, &next_page);
+      if (_ma_dispose(info, next_page.pos, 0))
+       goto err;
+
+      memcpy(leaf_buff, next_page.buff, (size_t) buff_length);
+      leaf_page->size= next_page.size;
+      leaf_page->flag= next_page.flag;
+
+      if (share->now_transactional)
+      {
+        /*
+          Log changes to parent page. Note that this page may have been
+          temporarily bigger than block_size.
+         */
+        if (_ma_log_delete(anc_page, key_deleted.key_pos,
+                           key_deleted.changed_length,
+                           key_deleted.move_length,
+                           anc_length - anc_page->org_size,
+                           KEY_OP_DEBUG_LOG_DEL_CHANGE_2))
+          goto err;
+        /*
+          Log changes to leaf page. Data for leaf page is in leaf_buff
+          which contains original leaf_buff, parting key and next_buff
+        */
+        if (_ma_log_suffix(leaf_page, leaf_length, buff_length))
+          goto err;
+      }
+    }
+    else
+    {
+      /*
+        Balancing didn't free a page, so we have to split 'buff' into two
+        pages:
+        - Find key in middle of buffer
+        - Store everything before key in 'leaf_page'
+        - Pack key into anc_page at position of deleted key
+          Note that anc_page may overflow! (is handled by caller)
+        - Store remaining keys in next_page (buff)
+      */
+      MARIA_KEY_PARAM anc_key_inserted;
+
+      anc_end_pos= anc_buff + new_anc_length;
+
+      DBUG_PRINT("test",("anc_buff:%p anc_end_pos:%p",
+                         anc_buff, anc_end_pos));
+
+      if (!first_key && !_ma_get_last_key(&anc_key, anc_page, keypos))
+	goto err;
+      if (!(half_pos= _ma_find_half_pos(&leaf_key, &next_page, &after_key)))
+	goto err;
+      new_leaf_length= (uint) (half_pos - next_page.buff);
+      memcpy(leaf_buff, next_page.buff, (size_t) new_leaf_length);
+
+      leaf_page->size= new_leaf_length;
+      leaf_page->flag= page_flag;
+      page_store_info(share, leaf_page);
+
+      /* Correct new keypointer to leaf_page */
+      half_pos=after_key;
+      _ma_kpointer(info,
+                   leaf_key.data + leaf_key.data_length + leaf_key.ref_length,
+                   next_page.pos);
+
+      /* Save key in anc_page */
+      prev_key= (first_key  ? (uchar*) 0 : anc_key.data);
+      t_length= (*keyinfo->pack_key)(&leaf_key, key_reflength,
+                                     (keypos == anc_end_pos ? (uchar*) 0 :
+                                      keypos),
+                                     prev_key, prev_key, &anc_key_inserted);
+      if (t_length >= 0)
+	bmove_upp(anc_end_pos+t_length, anc_end_pos,
+                  (uint) (anc_end_pos - keypos));
+      else
+	bmove(keypos,keypos-t_length,(uint) (anc_end_pos-keypos)+t_length);
+      (*keyinfo->store_key)(keyinfo,keypos, &anc_key_inserted);
+      new_anc_length+= t_length;
+      anc_page->size= new_anc_length;
+      page_store_size(share, anc_page);
+
+      if (leaf_key.flag & (SEARCH_USER_KEY_HAS_TRANSID |
+                           SEARCH_PAGE_KEY_HAS_TRANSID))
+        _ma_mark_page_with_transid(share, anc_page);
+
+      /* Store key first in new page */
+      if (nod_flag)
+	bmove(next_page.buff + share->keypage_header, half_pos-nod_flag,
+              (size_t) nod_flag);
+      if (!(*keyinfo->get_key)(&leaf_key, page_flag, nod_flag, &half_pos))
+	goto err;
+      t_length=(int) (*keyinfo->pack_key)(&leaf_key, nod_flag, (uchar*) 0,
+					  (uchar*) 0, (uchar*) 0,
+					  &key_inserted);
+      /* t_length will always be > 0 for a new page !*/
+      tmp_length= (uint) ((next_page.buff + buff_length) - half_pos);
+      bmove(next_page.buff + p_length + t_length, half_pos, tmp_length);
+      (*keyinfo->store_key)(keyinfo, next_page.buff + p_length, &key_inserted);
+      new_buff_length= tmp_length + t_length + p_length;
+      next_page.size= new_buff_length;
+      page_store_size(share, &next_page);
+      /* keypage flag is already up to date */
+
+      if (share->now_transactional)
+      {
+        /*
+          Log changes to parent page
+          This has one key deleted from it and one key inserted to it at
+          keypos
+
+          ma_log_add ensures that we don't log changes that is outside of
+          key block size, as the REDO code can't handle that
+        */
+        if (_ma_log_add(anc_page, anc_length, keypos,
+                        anc_key_inserted.move_length +
+                        MY_MAX(anc_key_inserted.changed_length -
+                            anc_key_inserted.move_length,
+                            key_deleted.changed_length),
+                        anc_key_inserted.move_length -
+                        key_deleted.move_length, 1,
+                        KEY_OP_DEBUG_LOG_ADD_3))
+          goto err;
+
+        /*
+          Log changes to leaf page.
+          This contains original data with new data added at end
+        */
+        DBUG_ASSERT(leaf_length <= new_leaf_length);
+        if (_ma_log_suffix(leaf_page, leaf_length, new_leaf_length))
+          goto err;
+        /*
+          Log changes to next page
+
+          This contains original data with some prefix data deleted and
+          some compressed data at start possible extended
+
+          Data in buff was originally:
+          org_leaf_buff     [leaf_length]
+          separator_key     [buff_key_inserted.move_length]
+          next_key_changes  [buff_key_inserted.changed_length -move_length]
+          next_page_data    [next_buff_length - p_length -
+                            (buff_key_inserted.changed_length -move_length)]
+
+          After changes it's now:
+          unpacked_key      [key_inserted.changed_length]
+          next_suffix       [next_buff_length - key_inserted.changed_length]
+
+        */
+        DBUG_ASSERT(new_buff_length <= next_buff_length);
+        if (_ma_log_prefix(&next_page, key_inserted.changed_length,
+                           (int) (new_buff_length - next_buff_length),
+                           KEY_OP_DEBUG_LOG_PREFIX_1))
+          goto err;
+      }
+      page_mark_changed(info, &next_page);
+      if (_ma_write_keypage(&next_page,
+                            PAGECACHE_LOCK_LEFT_WRITELOCKED, DFLT_INIT_HITS))
+	goto err;
+    }
+
+    page_mark_changed(info, leaf_page);
+    if (_ma_write_keypage(leaf_page,
+                          PAGECACHE_LOCK_LEFT_WRITELOCKED, DFLT_INIT_HITS))
+      goto err;
+    stack_alloc_free(anc_key_buff, buff_alloced);
+    DBUG_RETURN(new_anc_length <=
+                ((info->quick_mode ? MARIA_MIN_KEYBLOCK_LENGTH :
+                  (uint) keyinfo->underflow_block_length)));
+  }
+
+  DBUG_PRINT("test",("use left page"));
+
+  keypos= _ma_get_last_key(&anc_key, anc_page, keypos);
+  if (!keypos)
+    goto err;
+  next_page.pos= _ma_kpos(key_reflength,keypos);
+  if (_ma_fetch_keypage(&next_page, info, keyinfo, next_page.pos,
+                        PAGECACHE_LOCK_WRITE, DFLT_INIT_HITS, info->buff, 0))
+    goto err;
+  buff_length= next_page.size;
+  endpos= next_page.buff + buff_length;
+  DBUG_DUMP("prev", next_page.buff, next_page.size);
+
+  /* find keys to make a big key-page */
+  bmove(next_keypos - key_reflength, leaf_buff + share->keypage_header,
+        key_reflength);
+  next_keypos=keypos;
+  if (!(*keyinfo->get_key)(&anc_key, anc_page_flag, key_reflength,
+                           &next_keypos))
+    goto err;
+  if (!_ma_get_last_key(&leaf_key, &next_page, endpos))
+    goto err;
+
+  /* merge pages and put parting key from anc_page between */
+  prev_key= (leaf_length == p_length ? (uchar*) 0 : leaf_key.data);
+  t_length=(*keyinfo->pack_key)(&anc_key, nod_flag,
+				(leaf_length == p_length ?
+                                 (uchar*) 0 : leaf_buff+p_length),
+				prev_key, prev_key,
+				&key_inserted);
+  if (t_length >= 0)
+    bmove(endpos+t_length, leaf_buff+p_length,
+          (size_t) (leaf_length-p_length));
+  else						/* We gained space */
+    bmove(endpos,leaf_buff+((int) p_length-t_length),
+	  (size_t) (leaf_length-p_length+t_length));
+  (*keyinfo->store_key)(keyinfo,endpos, &key_inserted);
+
+  /* Remember for logging how many bytes of leaf_buff that are not changed */
+  DBUG_ASSERT((int) key_inserted.changed_length >= key_inserted.move_length);
+  unchanged_leaf_length= (leaf_length - p_length -
+                          (key_inserted.changed_length -
+                           key_inserted.move_length));
+
+  new_buff_length= buff_length + leaf_length - p_length + t_length;
+
+#ifdef EXTRA_DEBUG
+  /* Ensure that unchanged_leaf_length is correct */
+  DBUG_ASSERT(bcmp(next_page.buff + new_buff_length - unchanged_leaf_length,
+                   leaf_buff + leaf_length - unchanged_leaf_length,
+                   unchanged_leaf_length) == 0);
+#endif
+
+  page_flag= next_page.flag | leaf_page->flag;
+  if (anc_key.flag & (SEARCH_USER_KEY_HAS_TRANSID |
+                       SEARCH_PAGE_KEY_HAS_TRANSID))
+    page_flag|= KEYPAGE_FLAG_HAS_TRANSID;
+
+  next_page.size= new_buff_length;
+  next_page.flag= page_flag;
+  page_store_info(share, &next_page);
+
+  /* remove key from anc_page */
+  if (!(s_length= remove_key(keyinfo, anc_page_flag, key_reflength, keypos,
+                             anc_key_buff,
+                             anc_buff+anc_length, (my_off_t *) 0,
+                             &key_deleted)))
+    goto err;
+
+  new_anc_length= anc_length - s_length;
+  anc_page->size= new_anc_length;
+  page_store_size(share, anc_page);
+
+  if (new_buff_length <= share->max_index_block_size)
+  {
+    /* All keys fitted into one page */
+    page_mark_changed(info, leaf_page);
+    if (_ma_dispose(info, leaf_page->pos, 0))
+      goto err;
+
+    if (share->now_transactional)
+    {
+      /*
+        Log changes to parent page. Note that this page may have been
+        temporarily bigger than block_size.
+      */
+      if (_ma_log_delete(anc_page, key_deleted.key_pos,
+                         key_deleted.changed_length, key_deleted.move_length,
+                         anc_length - anc_page->org_size,
+                         KEY_OP_DEBUG_LOG_DEL_CHANGE_3))
+        goto err;
+      /*
+        Log changes to next page. Data for leaf page is in buff
+        that contains original leaf_buff, parting key and next_buff
+      */
+      if (_ma_log_suffix(&next_page, buff_length, new_buff_length))
+        goto err;
+    }
+  }
+  else
+  {
+    /*
+      Balancing didn't free a page, so we have to split 'next_page' into two
+      pages
+      - Find key in middle of buffer (buff)
+      - Pack key at half_buff into anc_page at position of deleted key
+        Note that anc_page may overflow! (is handled by caller)
+      - Move everything after middlekey to 'leaf_buff'
+      - Shorten buff at 'endpos'
+    */
+    MARIA_KEY_PARAM anc_key_inserted;
+    size_t tmp_length;
+
+    if (keypos == anc_buff + share->keypage_header + key_reflength)
+      anc_pos= 0;				/* First key */
+    else
+    {
+      if (!_ma_get_last_key(&anc_key, anc_page, keypos))
+        goto err;
+      anc_pos= anc_key.data;
+    }
+    if (!(endpos= _ma_find_half_pos(&leaf_key, &next_page, &half_pos)))
+      goto err;
+
+    /* Correct new keypointer to leaf_page */
+    _ma_kpointer(info,leaf_key.data + leaf_key.data_length +
+                 leaf_key.ref_length, leaf_page->pos);
+
+    /* Save parting key found by _ma_find_half_pos() in anc_page */
+    DBUG_DUMP("anc_buff", anc_buff, new_anc_length);
+    DBUG_DUMP_KEY("key_to_anc", &leaf_key);
+    anc_end_pos= anc_buff + new_anc_length;
+    t_length=(*keyinfo->pack_key)(&leaf_key, key_reflength,
+				  keypos == anc_end_pos ? (uchar*) 0
+				  : keypos,
+				  anc_pos, anc_pos,
+				  &anc_key_inserted);
+    if (t_length >= 0)
+      bmove_upp(anc_end_pos+t_length, anc_end_pos,
+                (uint) (anc_end_pos-keypos));
+    else
+      bmove(keypos,keypos-t_length,(uint) (anc_end_pos-keypos)+t_length);
+    (*keyinfo->store_key)(keyinfo,keypos, &anc_key_inserted);
+    new_anc_length+= t_length;
+    anc_page->size= new_anc_length;
+    page_store_size(share, anc_page);
+
+    if (leaf_key.flag & (SEARCH_USER_KEY_HAS_TRANSID |
+                         SEARCH_PAGE_KEY_HAS_TRANSID))
+      _ma_mark_page_with_transid(share, anc_page);
+
+    /* Store first key on new page */
+    if (nod_flag)
+      bmove(leaf_buff + share->keypage_header, half_pos-nod_flag,
+            (size_t) nod_flag);
+    if (!(*keyinfo->get_key)(&leaf_key, page_flag, nod_flag, &half_pos))
+      goto err;
+    DBUG_DUMP_KEY("key_to_leaf", &leaf_key);
+    t_length=(*keyinfo->pack_key)(&leaf_key, nod_flag, (uchar*) 0,
+				  (uchar*) 0, (uchar*) 0, &key_inserted);
+    /* t_length will always be > 0 for a new page !*/
+    tmp_length= (size_t) ((next_page.buff + new_buff_length) - half_pos);
+    DBUG_PRINT("info",("t_length: %d  length: %d",t_length, (int) tmp_length));
+    bmove(leaf_buff+p_length+t_length, half_pos, tmp_length);
+    (*keyinfo->store_key)(keyinfo,leaf_buff+p_length, &key_inserted);
+    new_leaf_length= (uint)(tmp_length + t_length + p_length);
+    DBUG_ASSERT(new_leaf_length <= share->max_index_block_size);
+
+    leaf_page->size= new_leaf_length;
+    leaf_page->flag= page_flag;
+    page_store_info(share, leaf_page);
+
+    new_buff_length= (uint) (endpos - next_page.buff);
+    next_page.size= new_buff_length;
+    page_store_size(share, &next_page);
+
+    if (share->now_transactional)
+    {
+      /*
+        Log changes to parent page
+        This has one key deleted from it and one key inserted to it at
+        keypos
+
+        ma_log_add() ensures that we don't log changes that is outside of
+        key block size, as the REDO code can't handle that
+      */
+      if (_ma_log_add(anc_page, anc_length, keypos,
+                      anc_key_inserted.move_length +
+                      MY_MAX(anc_key_inserted.changed_length -
+                          anc_key_inserted.move_length,
+                          key_deleted.changed_length),
+                      anc_key_inserted.move_length -
+                      key_deleted.move_length, 1,KEY_OP_DEBUG_LOG_ADD_4))
+        goto err;
+
+      /*
+        Log changes to leaf page.
+        This contains original data with new data added first
+      */
+      DBUG_ASSERT(leaf_length <= new_leaf_length);
+      DBUG_ASSERT(new_leaf_length >= unchanged_leaf_length);
+      if (_ma_log_prefix(leaf_page, new_leaf_length - unchanged_leaf_length,
+                         (int) (new_leaf_length - leaf_length),
+                         KEY_OP_DEBUG_LOG_PREFIX_2))
+        goto err;
+      /*
+        Log changes to next page
+        This contains original data with some suffix data deleted
+      */
+      DBUG_ASSERT(new_buff_length <= buff_length);
+      if (_ma_log_suffix(&next_page, buff_length, new_buff_length))
+        goto err;
+    }
+
+    page_mark_changed(info, leaf_page);
+    if (_ma_write_keypage(leaf_page,
+                          PAGECACHE_LOCK_LEFT_WRITELOCKED, DFLT_INIT_HITS))
+      goto err;
+  }
+  page_mark_changed(info, &next_page);
+  if (_ma_write_keypage(&next_page,
+                        PAGECACHE_LOCK_LEFT_WRITELOCKED, DFLT_INIT_HITS))
+    goto err;
+
+  stack_alloc_free(anc_key_buff, buff_alloced);
+  DBUG_RETURN(new_anc_length <=
+              ((info->quick_mode ? MARIA_MIN_KEYBLOCK_LENGTH :
+                (uint) keyinfo->underflow_block_length)));
+
+err:
+  stack_alloc_free(anc_key_buff, buff_alloced);
+  DBUG_RETURN(-1);
+} /* underflow */
+
+
+/**
+  @brief Remove a key from page
+
+  @fn remove_key()
+    keyinfo	          Key handle
+    nod_flag              Length of node ptr
+    keypos	          Where on page key starts
+    lastkey	          Buffer for storing keys to be removed
+    page_end	          Pointer to end of page
+    next_block	          If <> 0 and node-page, this is set to address of
+    		          next page
+    s_temp	          Information about what changes was done one the page:
+    s_temp.key_pos        Start of key
+    s_temp.move_length    Number of bytes removed at keypos
+    s_temp.changed_length Number of bytes changed at keypos
+
+  @todo
+    The current code doesn't handle the case that the next key may be
+    packed better against the previous key if there is a case difference
+
+  @return
+  @retval 0  error
+  @retval #  How many chars was removed
+*/
+
+static uint remove_key(MARIA_KEYDEF *keyinfo, uint page_flag, uint nod_flag,
+		       uchar *keypos, uchar *lastkey,
+		       uchar *page_end, my_off_t *next_block,
+                       MARIA_KEY_PARAM *s_temp)
+{
+  int s_length;
+  uchar *start;
+  DBUG_ENTER("remove_key");
+  DBUG_PRINT("enter", ("keypos:%p page_end: %p",
+                       keypos, page_end));
+
+  start= s_temp->key_pos= keypos;
+  s_temp->changed_length= 0;
+  if (!(keyinfo->flag &
+	(HA_PACK_KEY | HA_SPACE_PACK_USED | HA_VAR_LENGTH_KEY |
+	 HA_BINARY_PACK_KEY)) &&
+      !(page_flag & KEYPAGE_FLAG_HAS_TRANSID))
+  {
+    /* Static length key */
+    s_length=(int) (keyinfo->keylength+nod_flag);
+    if (next_block && nod_flag)
+      *next_block= _ma_kpos(nod_flag,keypos+s_length);
+  }
+  else
+  {
+    /* Let keypos point at next key */
+    MARIA_KEY key;
+
+    /* Calculate length of key */
+    key.keyinfo= keyinfo;
+    key.data=    lastkey;
+    if (!(*keyinfo->get_key)(&key, page_flag, nod_flag, &keypos))
+      DBUG_RETURN(0);				/* Error */
+
+    if (next_block && nod_flag)
+      *next_block= _ma_kpos(nod_flag,keypos);
+    s_length=(int) (keypos-start);
+    if (keypos != page_end)
+    {
+      if (keyinfo->flag & HA_BINARY_PACK_KEY)
+      {
+	uchar *old_key= start;
+	uint next_length,prev_length,prev_pack_length;
+
+        /* keypos points here on start of next key */
+	get_key_length(next_length,keypos);
+	get_key_pack_length(prev_length,prev_pack_length,old_key);
+	if (next_length > prev_length)
+	{
+          uint diff= (next_length-prev_length);
+	  /* We have to copy data from the current key to the next key */
+	  keypos-= diff + prev_pack_length;
+	  store_key_length(keypos, prev_length);
+          bmove(keypos + prev_pack_length, lastkey + prev_length, diff);
+	  s_length=(int) (keypos-start);
+          s_temp->changed_length= diff + prev_pack_length;
+	}
+      }
+      else
+      {
+	/* Check if a variable length first key part */
+	if ((keyinfo->seg->flag & HA_PACK_KEY) && *keypos & 128)
+	{
+	  /* Next key is packed against the current one */
+	  uint next_length,prev_length,prev_pack_length,lastkey_length,
+	    rest_length;
+	  if (keyinfo->seg[0].length >= 127)
+	  {
+	    if (!(prev_length=mi_uint2korr(start) & 32767))
+	      goto end;
+	    next_length=mi_uint2korr(keypos) & 32767;
+	    keypos+=2;
+	    prev_pack_length=2;
+	  }
+	  else
+	  {
+	    if (!(prev_length= *start & 127))
+	      goto end;				/* Same key as previous*/
+	    next_length= *keypos & 127;
+	    keypos++;
+	    prev_pack_length=1;
+	  }
+	  if (!(*start & 128))
+	    prev_length=0;			/* prev key not packed */
+	  if (keyinfo->seg[0].flag & HA_NULL_PART)
+	    lastkey++;				/* Skip null marker */
+	  get_key_length(lastkey_length,lastkey);
+	  if (!next_length)			/* Same key after */
+	  {
+	    next_length=lastkey_length;
+	    rest_length=0;
+	  }
+	  else
+	    get_key_length(rest_length,keypos);
+
+	  if (next_length >= prev_length)
+	  {
+            /* Next key is based on deleted key */
+            uint pack_length;
+            uint diff= (next_length-prev_length);
+
+            /* keypos points to data of next key (after key length) */
+	    bmove(keypos - diff, lastkey + prev_length, diff);
+	    rest_length+= diff;
+	    pack_length= prev_length ? get_pack_length(rest_length): 0;
+	    keypos-= diff + pack_length + prev_pack_length;
+	    s_length=(int) (keypos-start);
+	    if (prev_length)			/* Pack against prev key */
+	    {
+	      *keypos++= start[0];
+	      if (prev_pack_length == 2)
+		*keypos++= start[1];
+	      store_key_length(keypos,rest_length);
+	    }
+	    else
+	    {
+	      /* Next key is not packed anymore */
+	      if (keyinfo->seg[0].flag & HA_NULL_PART)
+	      {
+		rest_length++;			/* Mark not null */
+	      }
+	      if (prev_pack_length == 2)
+	      {
+		mi_int2store(keypos,rest_length);
+	      }
+	      else
+		*keypos= rest_length;
+	    }
+            s_temp->changed_length= diff + pack_length + prev_pack_length;
+	  }
+	}
+      }
+    }
+  }
+  end:
+  bmove(start, start+s_length, (uint) (page_end-start-s_length));
+  s_temp->move_length= s_length;
+  DBUG_RETURN((uint) s_length);
+} /* remove_key */
+
+
+/****************************************************************************
+  Logging of redos
+****************************************************************************/
+
+/**
+   @brief
+   log entry where some parts are deleted and some things are changed
+   and some data could be added last.
+
+   @fn _ma_log_delete()
+   @param info		  Maria handler
+   @param page	          Pageaddress for changed page
+   @param buff		  Page buffer
+   @param key_pos         Start of change area
+   @param changed_length  How many bytes where changed at key_pos
+   @param move_length     How many bytes where deleted at key_pos
+   @param append_length	  Length of data added last
+		          This is taken from end of ma_page->buff
+
+   This is mainly used when a key is deleted. The append happens
+   when we delete a key from a page with data > block_size kept in
+   memory and we have to add back the data that was stored > block_size
+*/
+
+my_bool _ma_log_delete(MARIA_PAGE *ma_page, const uchar *key_pos,
+                       uint changed_length, uint move_length,
+                       uint append_length __attribute__((unused)),
+                       enum en_key_debug debug_marker __attribute__((unused)))
+{
+  LSN lsn;
+  uchar log_data[FILEID_STORE_SIZE + PAGE_STORE_SIZE + 2 + 5+ 2 + 3 + 3 + 6 + 3 + 7];
+  uchar *log_pos;
+  LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 7];
+  uint translog_parts, current_size, extra_length;
+  uint offset= (uint) (key_pos - ma_page->buff);
+  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_delete");
+  DBUG_PRINT("enter", ("page: %lu  offset: %u  changed_length: %u  move_length: %u  append_length: %u  page_size: %u",
+                       (ulong) page, offset, changed_length, move_length,
+                       append_length, ma_page->size));
+  DBUG_ASSERT(share->now_transactional && move_length);
+  DBUG_ASSERT(offset + changed_length <= ma_page->size);
+  DBUG_ASSERT(ma_page->org_size - move_length + append_length == ma_page->size);
+  DBUG_ASSERT(move_length <= ma_page->org_size - share->keypage_header);
+
+  /* Store address of new root page */
+  page_store(log_data + FILEID_STORE_SIZE, page);
+  log_pos= log_data+ FILEID_STORE_SIZE + PAGE_STORE_SIZE;
+  current_size= ma_page->org_size;
+
+#ifdef EXTRA_DEBUG_KEY_CHANGES
+  *log_pos++= KEY_OP_DEBUG;
+  *log_pos++= debug_marker;
+
+  *log_pos++= KEY_OP_DEBUG_2;
+  int2store(log_pos,   ma_page->org_size);
+  int2store(log_pos+2, ma_page->size);
+  log_pos+=4;
+#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_OFFSET;
+  int2store(log_pos+1, offset);
+  log_pos+= 3;
+  translog_parts= TRANSLOG_INTERNAL_PARTS + 1;
+  extra_length= 0;
+
+  if (changed_length)
+  {
+    if (offset + changed_length >= share->max_index_block_size)
+    {
+      changed_length= share->max_index_block_size - offset;
+      move_length= 0;                           /* Nothing to move */
+      current_size= share->max_index_block_size;
+    }
+
+    log_pos[0]= KEY_OP_CHANGE;
+    int2store(log_pos+1, changed_length);
+    log_pos+= 3;
+    log_array[translog_parts].str=    ma_page->buff + offset;
+    log_array[translog_parts].length= changed_length;
+    translog_parts++;
+
+    /* We only have to move things after offset+changed_length */
+    offset+= changed_length;
+  }
+
+  log_array[TRANSLOG_INTERNAL_PARTS + 0].str=    log_data;
+  log_array[TRANSLOG_INTERNAL_PARTS + 0].length= (uint) (log_pos - log_data);
+
+  if (move_length)
+  {
+    uint log_length;
+    if (offset + move_length < share->max_index_block_size)
+    {
+      /*
+        Move down things that is on page.
+        page_offset in apply_redo_inxed() will be at original offset
+        + changed_length.
+      */
+      log_pos[0]= KEY_OP_SHIFT;
+      int2store(log_pos+1, - (int) move_length);
+      log_length= 3;
+      current_size-= move_length;
+    }
+    else
+    {
+      /* Delete to end of page */
+      uint tmp= current_size - offset;
+      current_size= offset;
+      log_pos[0]= KEY_OP_DEL_SUFFIX;
+      int2store(log_pos+1, tmp);
+      log_length= 3;
+    }
+    log_array[translog_parts].str=    log_pos;
+    log_array[translog_parts].length= log_length;
+    translog_parts++;
+    log_pos+= log_length;
+    extra_length+= log_length;
+  }
+
+  if (current_size != ma_page->size &&
+      current_size != share->max_index_block_size)
+  {
+    /* Append data that didn't fit on the page before */
+    uint length= (MY_MIN(ma_page->size, share->max_index_block_size) -
+                  current_size);
+    uchar *data= ma_page->buff + current_size;
+
+    DBUG_ASSERT(length <= append_length);
+
+    log_pos[0]= KEY_OP_ADD_SUFFIX;
+    int2store(log_pos+1, length);
+    log_array[translog_parts].str=        log_pos;
+    log_array[translog_parts].length=     3;
+    log_array[translog_parts + 1].str=    data;
+    log_array[translog_parts + 1].length= length;
+    log_pos+= 3;
+    translog_parts+= 2;
+    current_size+= length;
+    extra_length+= 3 + length;
+  }
+
+  _ma_log_key_changes(ma_page,
+                      log_array + translog_parts,
+                      log_pos, &extra_length, &translog_parts);
+  /* Remember new page length for future log entires for same page */
+  ma_page->org_size= current_size;
+
+  if (translog_write_record(&lsn, LOGREC_REDO_INDEX,
+                            info->trn, info,
+                            (translog_size_t)
+                            log_array[TRANSLOG_INTERNAL_PARTS].length +
+                            changed_length + extra_length, translog_parts,
+                            log_array, log_data, NULL))
+    DBUG_RETURN(1);
+
+  DBUG_RETURN(0);
+}
+
+
+/****************************************************************************
+  Logging of undos
+****************************************************************************/
+
+my_bool _ma_write_undo_key_delete(MARIA_HA *info, const MARIA_KEY *key,
+                                  my_off_t new_root, LSN *res_lsn)
+{
+  MARIA_SHARE *share= info->s;
+  uchar log_data[LSN_STORE_SIZE + FILEID_STORE_SIZE +
+                 KEY_NR_STORE_SIZE + PAGE_STORE_SIZE], *log_pos;
+  LEX_CUSTRING log_array[TRANSLOG_INTERNAL_PARTS + 2];
+  struct st_msg_to_write_hook_for_undo_key msg;
+  enum translog_record_type log_type= LOGREC_UNDO_KEY_DELETE;
+  uint keynr= key->keyinfo->key_nr;
+
+  lsn_store(log_data, info->trn->undo_lsn);
+  key_nr_store(log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE, keynr);
+  log_pos= log_data + LSN_STORE_SIZE + FILEID_STORE_SIZE + KEY_NR_STORE_SIZE;
+
+  /**
+    @todo BUG if we had concurrent insert/deletes, reading state's key_root
+    like this would be unsafe.
+  */
+  if (new_root != share->state.key_root[keynr])
+  {
+    my_off_t page;
+    page= ((new_root == HA_OFFSET_ERROR) ? IMPOSSIBLE_PAGE_NO :
+           new_root / share->block_size);
+    page_store(log_pos, page);
+    log_pos+= PAGE_STORE_SIZE;
+    log_type= LOGREC_UNDO_KEY_DELETE_WITH_ROOT;
+  }
+
+  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=    key->data;
+  log_array[TRANSLOG_INTERNAL_PARTS + 1].length= (key->data_length +
+                                                  key->ref_length);
+
+  msg.root= &share->state.key_root[keynr];
+  msg.value= new_root;
+  /*
+    set autoincrement to 1 if this is an auto_increment key
+    This is only used if we are now in a rollback of a duplicate key
+  */
+  msg.auto_increment= share->base.auto_key == keynr + 1;
+
+  return translog_write_record(res_lsn, log_type,
+                               info->trn, info,
+                               (translog_size_t)
+                               (log_array[TRANSLOG_INTERNAL_PARTS + 0].length +
+                                log_array[TRANSLOG_INTERNAL_PARTS + 1].length),
+                               TRANSLOG_INTERNAL_PARTS + 2, log_array,
+                               log_data + LSN_STORE_SIZE, &msg) ? -1 : 0;
+}