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, 1063 insertions, 0 deletions

diff --git a/storage/myisam/mi_write.c b/storage/myisam/mi_write.c
new file mode 100644
index 00000000..e8a985a5
--- /dev/null
+++ b/storage/myisam/mi_write.c
@@ -0,0 +1,1063 @@
+/* Copyright (c) 2000, 2013, Oracle and/or its affiliates
+
+   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 St, Fifth Floor, Boston, MA 02110-1335  USA */
+
+/* Write a row to a MyISAM table */
+
+#include "fulltext.h"
+#include "rt_index.h"
+
+#define MAX_POINTER_LENGTH 8
+
+	/* Functions declared in this file */
+
+static int w_search(MI_INFO *info,MI_KEYDEF *keyinfo,
+		    uint comp_flag, uchar *key,
+		    uint key_length, my_off_t pos, uchar *father_buff,
+		    uchar *father_keypos, my_off_t father_page,
+		    my_bool insert_last);
+static int _mi_balance_page(MI_INFO *info,MI_KEYDEF *keyinfo,uchar *key,
+			    uchar *curr_buff,uchar *father_buff,
+			    uchar *father_keypos,my_off_t father_page);
+static uchar *_mi_find_last_pos(MI_KEYDEF *keyinfo, uchar *page,
+				uchar *key, uint *return_key_length,
+				uchar **after_key);
+int _mi_ck_write_tree(register MI_INFO *info, uint keynr,uchar *key,
+		      uint key_length);
+int _mi_ck_write_btree(register MI_INFO *info, uint keynr,uchar *key,
+		       uint key_length);
+
+	/* Write new record to database */
+
+int mi_write(MI_INFO *info, const uchar *record)
+{
+  MYISAM_SHARE *share=info->s;
+  uint i;
+  int save_errno;
+  my_off_t filepos;
+  uchar *buff;
+  my_bool lock_tree= share->concurrent_insert;
+  DBUG_ENTER("mi_write");
+  DBUG_PRINT("enter",("isam: %d  data: %d",info->s->kfile,info->dfile));
+
+  DBUG_EXECUTE_IF("myisam_pretend_crashed_table_on_usage",
+                  mi_print_error(info->s, HA_ERR_CRASHED);
+                  DBUG_RETURN(my_errno= HA_ERR_CRASHED););
+
+  /* it's always a bug to try to write a record with the deleted flag set */
+  DBUG_ASSERT(info->s->data_file_type != STATIC_RECORD || *record);
+
+  if (share->options & HA_OPTION_READ_ONLY_DATA)
+  {
+    DBUG_RETURN(my_errno=EACCES);
+  }
+  if (_mi_readinfo(info,F_WRLCK,1))
+    DBUG_RETURN(my_errno);
+
+  filepos= ((share->state.dellink != HA_OFFSET_ERROR &&
+             !info->append_insert_at_end) ?
+	    share->state.dellink :
+	    info->state->data_file_length);
+
+  if (share->base.reloc == (ha_rows) 1 &&
+      share->base.records == (ha_rows) 1 &&
+      info->state->records == (ha_rows) 1)
+  {						/* System file */
+    my_errno=HA_ERR_RECORD_FILE_FULL;
+    goto err2;
+  }
+  if (info->state->key_file_length >= share->base.margin_key_file_length)
+  {
+    my_errno=HA_ERR_INDEX_FILE_FULL;
+    goto err2;
+  }
+  if (_mi_mark_file_changed(info))
+    goto err2;
+
+  /* Calculate and check all unique constraints */
+  for (i=0 ; i < share->state.header.uniques ; i++)
+  {
+    MI_UNIQUEDEF *def= share->uniqueinfo + i;
+    if (mi_is_key_active(share->state.key_map, def->key) &&
+        mi_check_unique(info, def, record, mi_unique_hash(def, record),
+                        HA_OFFSET_ERROR))
+      goto err2;
+  }
+
+	/* Write all keys to indextree */
+
+  buff=info->lastkey2;
+  for (i=0 ; i < share->base.keys ; i++)
+  {
+    if (mi_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(&share->key_root_lock[i]);
+	share->keyinfo[i].version++;
+      }
+      if (share->keyinfo[i].flag & HA_FULLTEXT )
+      {
+        if (_mi_ft_add(info,i, buff, record, filepos))
+        {
+	  if (local_lock_tree)
+            mysql_rwlock_unlock(&share->key_root_lock[i]);
+          DBUG_PRINT("error",("Got error: %d on write",my_errno));
+          goto err;
+        }
+      }
+      else
+      {
+        if (share->keyinfo[i].ck_insert(info,i,buff,
+			_mi_make_key(info,i,buff,record,filepos)))
+        {
+          if (local_lock_tree)
+            mysql_rwlock_unlock(&share->key_root_lock[i]);
+          DBUG_PRINT("error",("Got error: %d on write",my_errno));
+          goto err;
+        }
+      }
+
+      /* The above changed info->lastkey2. Inform mi_rnext_same(). */
+      info->update&= ~HA_STATE_RNEXT_SAME;
+
+      if (local_lock_tree)
+        mysql_rwlock_unlock(&share->key_root_lock[i]);
+    }
+  }
+  if (share->calc_checksum)
+    info->checksum=(*share->calc_checksum)(info,record);
+  if (!(info->opt_flag & OPT_NO_ROWS))
+  {
+    if ((*share->write_record)(info,record))
+      goto err;
+    info->state->checksum+=info->checksum;
+  }
+  if (share->base.auto_key)
+    set_if_bigger(info->s->state.auto_increment,
+                  retrieve_auto_increment(info, record));
+  info->update= (HA_STATE_CHANGED | HA_STATE_AKTIV | HA_STATE_WRITTEN |
+		 HA_STATE_ROW_CHANGED);
+  info->state->records++;
+  myisam_log_record(MI_LOG_WRITE,info,record,filepos,0);
+  (void) _mi_writeinfo(info, WRITEINFO_UPDATE_KEYFILE);
+  if (info->invalidator != 0)
+  {
+    DBUG_PRINT("info", ("invalidator... '%s' (update)", info->filename));
+    (*info->invalidator)(info->filename);
+    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)
+    mi_update_status((void*) info);
+
+  DBUG_RETURN(0);
+
+err:
+  save_errno=my_errno;
+  if (my_errno == HA_ERR_FOUND_DUPP_KEY || my_errno == HA_ERR_RECORD_FILE_FULL ||
+      my_errno == HA_ERR_NULL_IN_SPATIAL || my_errno == HA_ERR_OUT_OF_MEM)
+  {
+    if (info->bulk_insert)
+    {
+      uint j;
+      for (j=0 ; j < share->base.keys ; j++)
+        mi_flush_bulk_insert(info, j);
+    }
+    info->errkey= (int) i;
+    while ( i-- > 0)
+    {
+      if (mi_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(&share->key_root_lock[i]);
+	if (share->keyinfo[i].flag & HA_FULLTEXT)
+        {
+          if (_mi_ft_del(info,i, buff,record,filepos))
+	  {
+	    if (local_lock_tree)
+              mysql_rwlock_unlock(&share->key_root_lock[i]);
+            break;
+	  }
+        }
+        else
+	{
+	  uint key_length=_mi_make_key(info,i,buff,record,filepos);
+	  if (share->keyinfo[i].ck_delete(info, i, buff, key_length))
+	  {
+	    if (local_lock_tree)
+              mysql_rwlock_unlock(&share->key_root_lock[i]);
+	    break;
+	  }
+	}
+	if (local_lock_tree)
+          mysql_rwlock_unlock(&share->key_root_lock[i]);
+      }
+    }
+  }
+  else
+  {
+    mi_print_error(info->s, HA_ERR_CRASHED);
+    mi_mark_crashed(info);
+  }
+  info->update= (HA_STATE_CHANGED | HA_STATE_WRITTEN | HA_STATE_ROW_CHANGED);
+  my_errno=save_errno;
+err2:
+  save_errno=my_errno;
+  myisam_log_record(MI_LOG_WRITE,info,record,filepos,my_errno);
+  (void) _mi_writeinfo(info,WRITEINFO_UPDATE_KEYFILE);
+  DBUG_RETURN(my_errno=save_errno);
+} /* mi_write */
+
+
+	/* Write one key to btree */
+
+int _mi_ck_write(MI_INFO *info, uint keynr, uchar *key, uint key_length)
+{
+  DBUG_ENTER("_mi_ck_write");
+
+  if (info->bulk_insert && is_tree_inited(&info->bulk_insert[keynr]))
+  {
+    DBUG_RETURN(_mi_ck_write_tree(info, keynr, key, key_length));
+  }
+  else
+  {
+    DBUG_RETURN(_mi_ck_write_btree(info, keynr, key, key_length));
+  }
+} /* _mi_ck_write */
+
+
+/**********************************************************************
+ *                Normal insert code                                  *
+ **********************************************************************/
+
+int _mi_ck_write_btree(register MI_INFO *info, uint keynr, uchar *key,
+		       uint key_length)
+{
+  int error;
+  uint comp_flag;
+  MI_KEYDEF *keyinfo=info->s->keyinfo+keynr;
+  my_off_t  *root=&info->s->state.key_root[keynr];
+  DBUG_ENTER("_mi_ck_write_btree");
+
+  if (keyinfo->flag & HA_SORT_ALLOWS_SAME)
+    comp_flag=SEARCH_BIGGER;			/* Put after same key */
+  else if (keyinfo->flag & (HA_NOSAME|HA_FULLTEXT))
+  {
+    comp_flag=SEARCH_FIND | SEARCH_UPDATE | SEARCH_INSERT; /* No duplicates */
+    if (keyinfo->flag & HA_NULL_ARE_EQUAL)
+      comp_flag|= SEARCH_NULL_ARE_EQUAL;
+  }
+  else
+    comp_flag=SEARCH_SAME;			/* Keys in rec-pos order */
+
+  error=_mi_ck_real_write_btree(info, keyinfo, key, key_length,
+                                root, comp_flag);
+  if (info->ft1_to_ft2)
+  {
+    if (!error)
+      error= _mi_ft_convert_to_ft2(info, keynr, key);
+    delete_dynamic(info->ft1_to_ft2);
+    my_free(info->ft1_to_ft2);
+    info->ft1_to_ft2=0;
+  }
+  DBUG_RETURN(error);
+} /* _mi_ck_write_btree */
+
+int _mi_ck_real_write_btree(MI_INFO *info, MI_KEYDEF *keyinfo,
+    uchar *key, uint key_length, my_off_t *root, uint comp_flag)
+{
+  int error;
+  DBUG_ENTER("_mi_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, keyinfo, comp_flag, key, key_length,
+		      *root, (uchar *) 0, (uchar*) 0,
+		      (my_off_t) 0, 1)) > 0)
+    error=_mi_enlarge_root(info,keyinfo,key,root);
+  DBUG_RETURN(error);
+} /* _mi_ck_real_write_btree */
+
+
+	/* Make a new root with key as only pointer */
+
+int _mi_enlarge_root(MI_INFO *info, MI_KEYDEF *keyinfo, uchar *key,
+                     my_off_t *root)
+{
+  uint t_length,nod_flag;
+  MI_KEY_PARAM s_temp;
+  MYISAM_SHARE *share=info->s;
+  DBUG_ENTER("_mi_enlarge_root");
+
+  nod_flag= (*root != HA_OFFSET_ERROR) ?  share->base.key_reflength : 0;
+  _mi_kpointer(info,info->buff+2,*root); /* if nod */
+  t_length=(*keyinfo->pack_key)(keyinfo,nod_flag,(uchar*) 0,
+				(uchar*) 0, (uchar*) 0, key,&s_temp);
+  mi_putint(info->buff,t_length+2+nod_flag,nod_flag);
+  (*keyinfo->store_key)(keyinfo,info->buff+2+nod_flag,&s_temp);
+  info->buff_used=info->page_changed=1;		/* info->buff is used */
+  if ((*root= _mi_new(info,keyinfo,DFLT_INIT_HITS)) == HA_OFFSET_ERROR ||
+      _mi_write_keypage(info,keyinfo,*root,DFLT_INIT_HITS,info->buff))
+    DBUG_RETURN(-1);
+  DBUG_RETURN(0);
+} /* _mi_enlarge_root */
+
+
+	/*
+	  Search after a position for a key and store it there
+	  Returns -1 = error
+		   0  = ok
+		   1  = key should be stored in higher tree
+	*/
+
+static int w_search(register MI_INFO *info, register MI_KEYDEF *keyinfo,
+		    uint comp_flag, uchar *key, uint key_length, my_off_t page,
+		    uchar *father_buff, uchar *father_keypos,
+		    my_off_t father_page, my_bool insert_last)
+{
+  int error,flag;
+  uint nod_flag, search_key_length;
+  uchar *temp_buff,*keypos;
+  uchar keybuff[HA_MAX_KEY_BUFF];
+  my_bool was_last_key;
+  my_off_t next_page, dupp_key_pos;
+  DBUG_ENTER("w_search");
+  DBUG_PRINT("enter",("page: %ld", (long) page));
+
+  search_key_length= (comp_flag & SEARCH_FIND) ? key_length : USE_WHOLE_KEY;
+  if (!(temp_buff= (uchar*) my_alloca((uint) keyinfo->block_length+
+				      HA_MAX_KEY_BUFF*2)))
+    DBUG_RETURN(-1);
+  if (!_mi_fetch_keypage(info,keyinfo,page,DFLT_INIT_HITS,temp_buff,0))
+    goto err;
+
+  flag=(*keyinfo->bin_search)(info,keyinfo,temp_buff,key,search_key_length,
+			      comp_flag, &keypos, keybuff, &was_last_key);
+  nod_flag=mi_test_if_nod(temp_buff);
+  if (flag == 0)
+  {
+    uint tmp_key_length;
+	/* get position to record with duplicated key */
+    tmp_key_length=(*keyinfo->get_key)(keyinfo,nod_flag,&keypos,keybuff);
+    if (tmp_key_length)
+      dupp_key_pos=_mi_dpos(info,0,keybuff+tmp_key_length);
+    else
+      dupp_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)(info, keyinfo, temp_buff, key,
+                                    USE_WHOLE_KEY, comp_flag,
+                                    &keypos, keybuff, &was_last_key);
+      }
+      else
+      {
+        /* popular word. two-level tree. going down */
+        my_off_t root=dupp_key_pos;
+        keyinfo=&info->s->ft2_keyinfo;
+        get_key_full_length_rdonly(off, key);
+        key+=off;
+        keypos-=keyinfo->keylength+nod_flag; /* we'll modify key entry 'in vivo' */
+        error=_mi_ck_real_write_btree(info, keyinfo, key, 0,
+                                      &root, comp_flag);
+        _mi_dpointer(info, keypos+HA_FT_WLEN, root);
+        subkeys--; /* should there be underflow protection ? */
+        DBUG_ASSERT(subkeys < 0);
+        ft_intXstore(keypos, subkeys);
+        if (!error)
+          error=_mi_write_keypage(info,keyinfo,page,DFLT_INIT_HITS,temp_buff);
+        my_afree((uchar*) temp_buff);
+        DBUG_RETURN(error);
+      }
+    }
+    else /* not HA_FULLTEXT, normal HA_NOSAME key */
+    {
+      info->dupp_key_pos= dupp_key_pos;
+      my_afree((uchar*) temp_buff);
+      my_errno=HA_ERR_FOUND_DUPP_KEY;
+      DBUG_RETURN(-1);
+    }
+  }
+  if (flag == MI_FOUND_WRONG_KEY)
+    DBUG_RETURN(-1);
+  if (!was_last_key)
+    insert_last=0;
+  next_page=_mi_kpos(nod_flag,keypos);
+  if (next_page == HA_OFFSET_ERROR ||
+      (error=w_search(info, keyinfo, comp_flag, key, key_length, next_page,
+		      temp_buff, keypos, page, insert_last)) >0)
+  {
+    error=_mi_insert(info,keyinfo,key,temp_buff,keypos,keybuff,father_buff,
+		     father_keypos,father_page, insert_last);
+    if (_mi_write_keypage(info,keyinfo,page,DFLT_INIT_HITS,temp_buff))
+      goto err;
+  }
+  my_afree((uchar*) temp_buff);
+  DBUG_RETURN(error);
+err:
+  my_afree((uchar*) temp_buff);
+  DBUG_PRINT("exit",("Error: %d",my_errno));
+  DBUG_RETURN (-1);
+} /* w_search */
+
+
+/*
+  Insert new key.
+
+  SYNOPSIS
+    _mi_insert()
+    info                        Open table information.
+    keyinfo                     Key definition information.
+    key                         New key.
+    anc_buff                    Key page (beginning).
+    key_pos                     Position in key page where to insert.
+    key_buff                    Copy of previous key.
+    father_buff                 parent key page for balancing.
+    father_key_pos              position in parent key page for balancing.
+    father_page                 position of parent key page in file.
+    insert_last                 If to append at end of page.
+
+  DESCRIPTION
+    Insert new key at right of key_pos.
+
+  RETURN
+    2           if key contains key to upper level.
+    0           OK.
+    < 0         Error.
+*/
+
+int _mi_insert(register MI_INFO *info, register MI_KEYDEF *keyinfo,
+	       uchar *key, uchar *anc_buff, uchar *key_pos, uchar *key_buff,
+               uchar *father_buff, uchar *father_key_pos, my_off_t father_page,
+	       my_bool insert_last)
+{
+  uint a_length,nod_flag;
+  int t_length;
+  uchar *endpos, *prev_key;
+  MI_KEY_PARAM s_temp;
+  DBUG_ENTER("_mi_insert");
+  DBUG_PRINT("enter",("key_pos: %p", key_pos));
+  DBUG_EXECUTE("key",_mi_print_key(DBUG_FILE,keyinfo->seg,key,USE_WHOLE_KEY););
+
+  nod_flag=mi_test_if_nod(anc_buff);
+  a_length=mi_getint(anc_buff);
+  endpos= anc_buff+ a_length;
+  prev_key=(key_pos == anc_buff+2+nod_flag ? (uchar*) 0 : key_buff);
+  t_length=(*keyinfo->pack_key)(keyinfo,nod_flag,
+				(key_pos == endpos ? (uchar*) 0 : key_pos),
+				prev_key, prev_key,
+				key,&s_temp);
+#ifndef DBUG_OFF
+  if (key_pos != anc_buff+2+nod_flag && (keyinfo->flag &
+					 (HA_BINARY_PACK_KEY | HA_PACK_KEY)))
+  {
+    DBUG_DUMP("prev_key",(uchar*) key_buff,_mi_keylength(keyinfo,key_buff));
+  }
+  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+MAX_POINTER_LENGTH)
+    {
+      mi_print_error(info->s, HA_ERR_CRASHED);
+      my_errno=HA_ERR_CRASHED;
+      DBUG_RETURN(-1);
+    }
+    bmove_upp((uchar*) endpos+t_length,(uchar*) endpos,(uint) (endpos-key_pos));
+  }
+  else
+  {
+    if (-t_length >= keyinfo->maxlength*2+MAX_POINTER_LENGTH)
+    {
+      mi_print_error(info->s, HA_ERR_CRASHED);
+      my_errno=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;
+  mi_putint(anc_buff,a_length,nod_flag);
+  if (a_length <= keyinfo->block_length)
+  {
+    if (keyinfo->block_length - a_length < 32 &&
+        keyinfo->flag & HA_FULLTEXT && key_pos == endpos &&
+        info->s->base.key_reflength <= info->s->rec_reflength &&
+        info->s->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
+       */
+      uchar *a=key, *b=anc_buff+2+nod_flag;
+      uint alen, blen, ft2len=info->s->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;
+#else
+      blen= *b++;
+#endif
+      get_key_length(alen,a);
+      DBUG_ASSERT(info->ft1_to_ft2==0);
+      if (alen == blen &&
+          ha_compare_word(keyinfo->seg->charset,
+                          a, alen,
+                          b, blen) == 0)
+      {
+        /* yup. converting */
+        info->ft1_to_ft2=(DYNAMIC_ARRAY *)
+          my_malloc(mi_key_memory_MI_INFO_ft1_to_ft2,
+                    sizeof(DYNAMIC_ARRAY), MYF(MY_WME));
+        my_init_dynamic_array(mi_key_memory_MI_INFO_ft1_to_ft2,
+                              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)
+          {
+            if (insert_dynamic(info->ft1_to_ft2, b))
+            {
+              mi_print_error(info->s, HA_ERR_OUT_OF_MEM);
+              my_errno= HA_ERR_OUT_OF_MEM;
+              DBUG_RETURN(-1);
+            }
+          }
+
+          /* fixing the page's length - it contains only one key now */
+          mi_putint(anc_buff,2+blen+ft2len+2,0);
+        }
+        /* the rest will be done when we're back from recursion */
+      }
+    }
+    DBUG_RETURN(0);				/* There is room on page */
+  }
+  /* Page is full */
+  if (nod_flag)
+    insert_last=0;
+  if (!(keyinfo->flag & (HA_VAR_LENGTH_KEY | HA_BINARY_PACK_KEY)) &&
+      father_buff && !insert_last)
+    DBUG_RETURN(_mi_balance_page(info,keyinfo,key,anc_buff,father_buff,
+				 father_key_pos,father_page));
+  DBUG_RETURN(_mi_split_page(info,keyinfo,key,anc_buff,key_buff, insert_last));
+} /* _mi_insert */
+
+
+	/* split a full page in two and assign emerging item to key */
+
+int _mi_split_page(register MI_INFO *info, register MI_KEYDEF *keyinfo,
+		   uchar *key, uchar *buff, uchar *key_buff,
+		   my_bool insert_last_key)
+{
+  uint length,a_length,key_ref_length,t_length,nod_flag,key_length;
+  uchar *key_pos,*pos, *UNINIT_VAR(after_key);
+  my_off_t new_pos;
+  MI_KEY_PARAM s_temp;
+  DBUG_ENTER("mi_split_page");
+  DBUG_DUMP("buff",(uchar*) buff,mi_getint(buff));
+
+  if (info->s->keyinfo+info->lastinx == keyinfo)
+    info->page_changed=1;			/* Info->buff is used */
+  info->buff_used=1;
+  nod_flag=mi_test_if_nod(buff);
+  key_ref_length=2+nod_flag;
+  if (insert_last_key)
+    key_pos=_mi_find_last_pos(keyinfo,buff,key_buff, &key_length, &after_key);
+  else
+    key_pos=_mi_find_half_pos(nod_flag,keyinfo,buff,key_buff, &key_length,
+			      &after_key);
+  if (!key_pos)
+    DBUG_RETURN(-1);
+
+  length=(uint) (key_pos-buff);
+  a_length=mi_getint(buff);
+  mi_putint(buff,length,nod_flag);
+
+  key_pos=after_key;
+  if (nod_flag)
+  {
+    DBUG_PRINT("test",("Splitting nod"));
+    pos=key_pos-nod_flag;
+    memcpy((uchar*) info->buff+2,(uchar*) pos,(size_t) nod_flag);
+  }
+
+	/* Move middle item to key and pointer to new page */
+  if ((new_pos=_mi_new(info,keyinfo,DFLT_INIT_HITS)) == HA_OFFSET_ERROR)
+    DBUG_RETURN(-1);
+  _mi_kpointer(info,_mi_move_key(keyinfo,key,key_buff),new_pos);
+
+	/* Store new page */
+  if (!(*keyinfo->get_key)(keyinfo,nod_flag,&key_pos,key_buff))
+    DBUG_RETURN(-1);
+
+  t_length=(*keyinfo->pack_key)(keyinfo,nod_flag,(uchar *) 0,
+				(uchar*) 0, (uchar*) 0,
+				key_buff, &s_temp);
+  length=(uint) ((buff+a_length)-key_pos);
+  memcpy((uchar*) info->buff+key_ref_length+t_length,(uchar*) key_pos,
+	 (size_t) length);
+  (*keyinfo->store_key)(keyinfo,info->buff+key_ref_length,&s_temp);
+  mi_putint(info->buff,length+t_length+key_ref_length,nod_flag);
+
+  if (_mi_write_keypage(info,keyinfo,new_pos,DFLT_INIT_HITS,info->buff))
+    DBUG_RETURN(-1);
+  DBUG_DUMP("key",(uchar*) key,_mi_keylength(keyinfo,key));
+  DBUG_RETURN(2);				/* Middle key up */
+} /* _mi_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.
+	  return_key_length will contain the length of key
+	  after_key will contain the position to where the next key starts
+	*/
+
+uchar *_mi_find_half_pos(uint nod_flag, MI_KEYDEF *keyinfo, uchar *page,
+			 uchar *key, uint *return_key_length,
+			 uchar **after_key)
+{
+  uint keys,length,key_ref_length;
+  uchar *end,*lastpos;
+  DBUG_ENTER("_mi_find_half_pos");
+
+  key_ref_length=2+nod_flag;
+  length=mi_getint(page)-key_ref_length;
+  page+=key_ref_length;
+  if (!(keyinfo->flag &
+	(HA_PACK_KEY | HA_SPACE_PACK_USED | HA_VAR_LENGTH_KEY |
+	 HA_BINARY_PACK_KEY)))
+  {
+    key_ref_length=keyinfo->keylength+nod_flag;
+    keys=length/(key_ref_length*2);
+    *return_key_length=keyinfo->keylength;
+    end=page+keys*key_ref_length;
+    *after_key=end+key_ref_length;
+    memcpy(key,end,key_ref_length);
+    DBUG_RETURN(end);
+  }
+
+  end=page+length/2-key_ref_length;		/* This is aprox. half */
+  *key='\0';
+  do
+  {
+    lastpos=page;
+    if (!(length=(*keyinfo->get_key)(keyinfo,nod_flag,&page,key)))
+      DBUG_RETURN(0);
+  } while (page < end);
+  *return_key_length=length;
+  *after_key=page;
+  DBUG_PRINT("exit",("returns: %p  page: %p  half: %p",
+                     lastpos, page, end));
+  DBUG_RETURN(lastpos);
+} /* _mi_find_half_pos */
+
+
+/*
+  Split buffer at last key
+  Returns pointer to the start of the key before the last key
+  key will contain the last key
+*/
+
+static uchar *_mi_find_last_pos(MI_KEYDEF *keyinfo, uchar *page,
+				uchar *key, uint *return_key_length,
+				uchar **after_key)
+{
+  uint keys, length, UNINIT_VAR(last_length), key_ref_length;
+  uchar *end,*lastpos,*prevpos;
+  uchar key_buff[HA_MAX_KEY_BUFF];
+  DBUG_ENTER("_mi_find_last_pos");
+
+  key_ref_length=2;
+  length=mi_getint(page)-key_ref_length;
+  page+=key_ref_length;
+  if (!(keyinfo->flag &
+	(HA_PACK_KEY | HA_SPACE_PACK_USED | HA_VAR_LENGTH_KEY |
+	 HA_BINARY_PACK_KEY)))
+  {
+    keys=length/keyinfo->keylength-2;
+    *return_key_length=length=keyinfo->keylength;
+    end=page+keys*length;
+    *after_key=end+length;
+    memcpy(key,end,length);
+    DBUG_RETURN(end);
+  }
+
+  end=page+length-key_ref_length;
+  DBUG_ASSERT(page < end);
+  *key='\0';
+  length=0;
+  lastpos=page;
+
+  do
+  {
+    prevpos=lastpos; lastpos=page;
+    last_length=length;
+    memcpy(key, key_buff, length);		/* previous key */
+    if (!(length=(*keyinfo->get_key)(keyinfo,0,&page,key_buff)))
+    {
+      mi_print_error(keyinfo->share, HA_ERR_CRASHED);
+      my_errno=HA_ERR_CRASHED;
+      DBUG_RETURN(0);
+    }
+  } while (page < end);
+
+  *return_key_length=last_length;
+  *after_key=lastpos;
+  DBUG_PRINT("exit",("returns: %p  page: %p  end: %p",
+                     prevpos, page, end));
+  DBUG_RETURN(prevpos);
+} /* _mi_find_last_pos */
+
+
+	/* Balance page with not packed keys with page on right/left */
+	/* returns 0 if balance was done */
+
+static int _mi_balance_page(register MI_INFO *info, MI_KEYDEF *keyinfo,
+			    uchar *key, uchar *curr_buff, uchar *father_buff,
+			    uchar *father_key_pos, my_off_t father_page)
+{
+  my_bool right;
+  uint k_length,father_length,father_keylength,nod_flag,curr_keylength,
+       right_length,left_length,new_right_length,new_left_length,extra_length,
+       length,keys;
+  uchar *pos,*buff,*extra_buff;
+  my_off_t next_page,new_pos;
+  uchar tmp_part_key[HA_MAX_KEY_BUFF];
+  DBUG_ENTER("_mi_balance_page");
+
+  k_length=keyinfo->keylength;
+  father_length=mi_getint(father_buff);
+  father_keylength=k_length+info->s->base.key_reflength;
+  nod_flag=mi_test_if_nod(curr_buff);
+  curr_keylength=k_length+nod_flag;
+  info->page_changed=1;
+
+  if ((father_key_pos != father_buff+father_length &&
+       (info->state->records & 1)) ||
+      father_key_pos == father_buff+2+info->s->base.key_reflength)
+  {
+    right=1;
+    next_page= _mi_kpos(info->s->base.key_reflength,
+			father_key_pos+father_keylength);
+    buff=info->buff;
+    DBUG_PRINT("test",("use right page: %lu", (ulong) next_page));
+  }
+  else
+  {
+    right=0;
+    father_key_pos-=father_keylength;
+    next_page= _mi_kpos(info->s->base.key_reflength,father_key_pos);
+					/* Fix that curr_buff is to left */
+    buff=curr_buff; curr_buff=info->buff;
+    DBUG_PRINT("test",("use left page: %lu", (ulong) next_page));
+  }					/* father_key_pos ptr to parting key */
+
+  if (!_mi_fetch_keypage(info,keyinfo,next_page,DFLT_INIT_HITS,info->buff,0))
+    goto err;
+  DBUG_DUMP("next",(uchar*) info->buff,mi_getint(info->buff));
+
+	/* Test if there is room to share keys */
+
+  left_length=mi_getint(curr_buff);
+  right_length=mi_getint(buff);
+  keys=(left_length+right_length-4-nod_flag*2)/curr_keylength;
+
+  if ((right ? right_length : left_length) + curr_keylength <=
+      keyinfo->block_length)
+  {						/* Merge buffs */
+    new_left_length=2+nod_flag+(keys/2)*curr_keylength;
+    new_right_length=2+nod_flag+((keys+1)/2)*curr_keylength;
+    mi_putint(curr_buff,new_left_length,nod_flag);
+    mi_putint(buff,new_right_length,nod_flag);
+
+    if (left_length < new_left_length)
+    {						/* Move keys buff -> leaf */
+      pos=curr_buff+left_length;
+      memcpy((uchar*) pos,(uchar*) father_key_pos, (size_t) k_length);
+      memcpy((uchar*) pos+k_length, (uchar*) buff+2,
+	     (size_t) (length=new_left_length - left_length - k_length));
+      pos=buff+2+length;
+      memcpy((uchar*) father_key_pos,(uchar*) pos,(size_t) k_length);
+      bmove((uchar*) buff + 2, (uchar*) pos + k_length, new_right_length - 2);
+    }
+    else
+    {						/* Move keys -> buff */
+
+      bmove_upp((uchar*) buff+new_right_length,(uchar*) buff+right_length,
+		right_length-2);
+      length=new_right_length-right_length-k_length;
+      memcpy((uchar*) buff+2+length,father_key_pos,(size_t) k_length);
+      pos=curr_buff+new_left_length;
+      memcpy((uchar*) father_key_pos,(uchar*) pos,(size_t) k_length);
+      memcpy((uchar*) buff+2,(uchar*) pos+k_length,(size_t) length);
+    }
+
+    if (_mi_write_keypage(info,keyinfo,next_page,DFLT_INIT_HITS,info->buff) ||
+	_mi_write_keypage(info,keyinfo,father_page,DFLT_INIT_HITS,father_buff))
+      goto err;
+    DBUG_RETURN(0);
+  }
+
+	/* curr_buff[] and buff[] are full, lets split and make new nod */
+
+  extra_buff=info->buff+info->s->base.max_key_block_length;
+  new_left_length=new_right_length=2+nod_flag+(keys+1)/3*curr_keylength;
+  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;
+  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));
+  mi_putint(curr_buff,new_left_length,nod_flag);
+  mi_putint(buff,new_right_length,nod_flag);
+  mi_putint(extra_buff,extra_length+2,nod_flag);
+
+  /* move first largest keys to new page  */
+  pos=buff+right_length-extra_length;
+  memcpy((uchar*) extra_buff+2,pos,(size_t) extra_length);
+  /* Save new parting key */
+  memcpy(tmp_part_key, pos-k_length,k_length);
+  /* Make place for new keys */
+  bmove_upp((uchar*) buff+new_right_length,(uchar*) pos-k_length,
+	    right_length-extra_length-k_length-2);
+  /* Copy keys from left page */
+  pos= curr_buff+new_left_length;
+  memcpy((uchar*) buff+2,(uchar*) pos+k_length,
+	 (size_t) (length=left_length-new_left_length-k_length));
+  /* Copy old parting key */
+  memcpy((uchar*) buff+2+length,father_key_pos,(size_t) k_length);
+
+  /* Move new parting keys up to caller */
+  memcpy((uchar*) (right ? key : father_key_pos),pos,(size_t) k_length);
+  memcpy((uchar*) (right ? father_key_pos : key),tmp_part_key, k_length);
+
+  if ((new_pos=_mi_new(info,keyinfo,DFLT_INIT_HITS)) == HA_OFFSET_ERROR)
+    goto err;
+  _mi_kpointer(info,key+k_length,new_pos);
+  if (_mi_write_keypage(info,keyinfo,(right ? new_pos : next_page),
+			DFLT_INIT_HITS,info->buff) ||
+      _mi_write_keypage(info,keyinfo,(right ? next_page : new_pos),
+                        DFLT_INIT_HITS,extra_buff))
+    goto err;
+
+  DBUG_RETURN(1);				/* Middle key up */
+
+err:
+  DBUG_RETURN(-1);
+} /* _mi_balance_page */
+
+/**********************************************************************
+ *                Bulk insert code                                    *
+ **********************************************************************/
+
+typedef struct {
+  MI_INFO *info;
+  uint keynr;
+} bulk_insert_param;
+
+int _mi_ck_write_tree(register MI_INFO *info, uint keynr, uchar *key,
+		      uint key_length)
+{
+  int error;
+  DBUG_ENTER("_mi_ck_write_tree");
+
+  error= tree_insert(&info->bulk_insert[keynr], key,
+         key_length + info->s->rec_reflength,
+         info->bulk_insert[keynr].custom_arg) ? 0 : HA_ERR_OUT_OF_MEM ;
+
+  DBUG_RETURN(error);
+} /* _mi_ck_write_tree */
+
+
+/* typeof(_mi_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;
+  uchar lastkey[HA_MAX_KEY_BUFF], *key= (uchar*)key_arg;
+  uint keylen;
+  MI_KEYDEF *keyinfo;
+
+  switch (mode) {
+  case free_init:
+    if (param->info->s->concurrent_insert)
+    {
+      mysql_rwlock_wrlock(&param->info->s->key_root_lock[param->keynr]);
+      param->info->s->keyinfo[param->keynr].version++;
+    }
+    return 0;
+  case free_free:
+    keyinfo=param->info->s->keyinfo+param->keynr;
+    keylen=_mi_keylength(keyinfo, key);
+    memcpy(lastkey, key, keylen);
+    _mi_ck_write_btree(param->info, param->keynr, lastkey,
+			                 keylen - param->info->s->rec_reflength);
+    return 0;
+  case free_end:
+    if (param->info->s->concurrent_insert)
+      mysql_rwlock_unlock(&param->info->s->key_root_lock[param->keynr]);
+    return 0;
+  }
+  return 0;
+}
+
+
+int mi_init_bulk_insert(MI_INFO *info, size_t cache_size, ha_rows rows)
+{
+  MYISAM_SHARE *share=info->s;
+  MI_KEYDEF *key=share->keyinfo;
+  bulk_insert_param *params;
+  uint i, num_keys, total_keylength;
+  ulonglong key_map;
+  DBUG_ENTER("_mi_init_bulk_insert");
+  DBUG_PRINT("enter",("cache_size: %lu", (ulong) cache_size));
+
+  DBUG_ASSERT(!info->bulk_insert &&
+	      (!rows || rows >= MI_MIN_ROWS_TO_USE_BULK_INSERT));
+
+  mi_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) &&
+        mi_is_key_active(share->state.key_map, i))
+    {
+      num_keys++;
+      mi_set_key_active(key_map, i);
+      total_keylength+=key[i].maxlength+TREE_ELEMENT_EXTRA_SIZE;
+    }
+  }
+
+  if (num_keys==0 ||
+      num_keys * (size_t) MI_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(mi_key_memory_MI_INFO_bulk_insert,
+              (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 (mi_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 mi_flush_bulk_insert(MI_INFO *info, uint inx)
+{
+  if (info->bulk_insert)
+  {
+    if (is_tree_inited(&info->bulk_insert[inx]))
+      reset_tree(&info->bulk_insert[inx]);
+  }
+}
+
+int mi_end_bulk_insert(MI_INFO *info, my_bool abort)
+{
+  int first_error= 0;
+  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 ((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;
+  }
+  return first_error;
+}