Merge mysql.com:/home/mydev/mysql-5.1

into  mysql.com:/home/mydev/mysql-5.1-bug10405


storage/myisam/mi_check.c:
  Auto merged

19 files changed, 108 insertions, 63 deletions

diff --git a/storage/heap/hp_delete.c b/storage/heap/hp_delete.c
index 2d94418a1bf..f37db2588f3 100644
--- a/storage/heap/hp_delete.c
+++ b/storage/heap/hp_delete.c
@@ -79,7 +79,8 @@ int hp_rb_delete_key(HP_INFO *info, register HP_KEYDEF *keyinfo,
   custom_arg.key_length= hp_rb_make_key(keyinfo, info->recbuf, record, recpos);
   custom_arg.search_flag= SEARCH_SAME;
   old_allocated= keyinfo->rb_tree.allocated;
-  res= tree_delete(&keyinfo->rb_tree, info->recbuf, &custom_arg);
+  res= tree_delete(&keyinfo->rb_tree, info->recbuf, custom_arg.key_length,
+                   &custom_arg);
   info->s->index_length-= (old_allocated - keyinfo->rb_tree.allocated);
   return res;
 }
diff --git a/storage/myisam/ft_eval.c b/storage/myisam/ft_eval.c
index eab9d37af0b..459742d2aff 100644
--- a/storage/myisam/ft_eval.c
+++ b/storage/myisam/ft_eval.c
@@ -54,6 +54,7 @@ int main(int argc, char *argv[])
   /* Define a key over the first column */
   keyinfo[0].seg=keyseg;
   keyinfo[0].keysegs=1;
+  keyinfo[0].block_length= 0;                   /* Default block length */
   keyinfo[0].seg[0].type= HA_KEYTYPE_TEXT;
   keyinfo[0].seg[0].flag= HA_BLOB_PART;
   keyinfo[0].seg[0].start=recinfo[0].length;
diff --git a/storage/myisam/ft_test1.c b/storage/myisam/ft_test1.c
index d54b344e2cd..28dcfb2b758 100644
--- a/storage/myisam/ft_test1.c
+++ b/storage/myisam/ft_test1.c
@@ -89,6 +89,7 @@ static int run_test(const char *filename)
   /* Define a key over the first column */
   keyinfo[0].seg=keyseg;
   keyinfo[0].keysegs=1;
+  keyinfo[0].block_length= 0;                   /* Default block length */
   keyinfo[0].seg[0].type= key_type;
   keyinfo[0].seg[0].flag= (key_field == FIELD_BLOB) ? HA_BLOB_PART:
 			  (key_field == FIELD_VARCHAR) ? HA_VAR_LENGTH_PART:0;
diff --git a/storage/myisam/mi_check.c b/storage/myisam/mi_check.c
index 53fa4aa22c5..0fa095a21db 100644
--- a/storage/myisam/mi_check.c
+++ b/storage/myisam/mi_check.c
@@ -358,7 +358,7 @@ int chk_key(MI_CHECK *param, register MI_INFO *info)
     puts("- check key delete-chain");
 
   param->key_file_blocks=info->s->base.keystart;
-  for (key=0 ; key < info->s->state.header.max_block_size ; key++)
+  for (key=0 ; key < info->s->state.header.max_block_size_index ; key++)
     if (check_k_link(param,info,key))
     {
       if (param->testflag & T_VERBOSE) puts("");
@@ -1411,7 +1411,7 @@ int mi_repair(MI_CHECK *param, register MI_INFO *info,
     share->state.key_root[i]= HA_OFFSET_ERROR;
 
   /* Drop the delete chain. */
-  for (i=0 ; i < share->state.header.max_block_size ; i++)
+  for (i=0 ; i < share->state.header.max_block_size_index ; i++)
     share->state.key_del[i]=  HA_OFFSET_ERROR;
 
   /*
@@ -1795,7 +1795,7 @@ int mi_sort_index(MI_CHECK *param, register MI_INFO *info, my_string name)
   info->update= (short) (HA_STATE_CHANGED | HA_STATE_ROW_CHANGED);
   for (key=0 ; key < info->s->base.keys ; key++)
     info->s->state.key_root[key]=index_pos[key];
-  for (key=0 ; key < info->s->state.header.max_block_size ; key++)
+  for (key=0 ; key < info->s->state.header.max_block_size_index ; key++)
     info->s->state.key_del[key]=  HA_OFFSET_ERROR;
 
   info->s->state.changed&= ~STATE_NOT_SORTED_PAGES;
@@ -2095,7 +2095,7 @@ int mi_repair_by_sort(MI_CHECK *param, register MI_INFO *info,
     /* Clear the pointers to the given rows */
     for (i=0 ; i < share->base.keys ; i++)
       share->state.key_root[i]= HA_OFFSET_ERROR;
-    for (i=0 ; i < share->state.header.max_block_size ; i++)
+    for (i=0 ; i < share->state.header.max_block_size_index ; i++)
       share->state.key_del[i]=  HA_OFFSET_ERROR;
     info->state->key_file_length=share->base.keystart;
   }
@@ -2463,7 +2463,7 @@ int mi_repair_parallel(MI_CHECK *param, register MI_INFO *info,
     /* Clear the pointers to the given rows */
     for (i=0 ; i < share->base.keys ; i++)
       share->state.key_root[i]= HA_OFFSET_ERROR;
-    for (i=0 ; i < share->state.header.max_block_size ; i++)
+    for (i=0 ; i < share->state.header.max_block_size_index ; i++)
       share->state.key_del[i]=  HA_OFFSET_ERROR;
     info->state->key_file_length=share->base.keystart;
   }
diff --git a/storage/myisam/mi_create.c b/storage/myisam/mi_create.c
index 3be998b2c17..22cbde278be 100644
--- a/storage/myisam/mi_create.c
+++ b/storage/myisam/mi_create.c
@@ -28,9 +28,9 @@
 #endif
 #include <m_ctype.h>
 
-	/*
-	** Old options is used when recreating database, from isamchk
-	*/
+/*
+  Old options is used when recreating database, from myisamchk
+*/
 
 int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 	      uint columns, MI_COLUMNDEF *recinfo,
@@ -45,6 +45,7 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
        key_length,info_length,key_segs,options,min_key_length_skip,
        base_pos,long_varchar_count,varchar_length,
        max_key_block_length,unique_key_parts,fulltext_keys,offset;
+  uint aligned_key_start, block_length;
   ulong reclength, real_reclength,min_pack_length;
   char filename[FN_REFLEN],linkname[FN_REFLEN], *linkname_ptr;
   ulong pack_reclength;
@@ -428,8 +429,16 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
 	key_segs)
       share.state.rec_per_key_part[key_segs-1]=1L;
     length+=key_length;
+    /* Get block length for key, if defined by user */
+    block_length= (keydef->block_length ? 
+                   my_round_up_to_next_power(keydef->block_length) :
+                   myisam_block_size);
+    block_length= max(block_length, MI_MIN_KEY_BLOCK_LENGTH);
+    block_length= min(block_length, MI_MAX_KEY_BLOCK_LENGTH);
+
     keydef->block_length= MI_BLOCK_SIZE(length-real_length_diff,
-                                        pointer,MI_MAX_KEYPTR_SIZE);
+                                        pointer,MI_MAX_KEYPTR_SIZE,
+                                        block_length);
     if (keydef->block_length > MI_MAX_KEY_BLOCK_LENGTH ||
         length >= MI_MAX_KEY_BUFF)
     {
@@ -485,7 +494,7 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
   mi_int2store(share.state.header.base_pos,base_pos);
   share.state.header.language= (ci->language ?
 				ci->language : default_charset_info->number);
-  share.state.header.max_block_size=max_key_block_length/MI_MIN_KEY_BLOCK_LENGTH;
+  share.state.header.max_block_size_index= max_key_block_length/MI_MIN_KEY_BLOCK_LENGTH;
 
   share.state.dellink = HA_OFFSET_ERROR;
   share.state.process=	(ulong) getpid();
@@ -512,8 +521,12 @@ int mi_create(const char *name,uint keys,MI_KEYDEF *keydefs,
   mi_int2store(share.state.header.unique_key_parts,unique_key_parts);
 
   mi_set_all_keys_active(share.state.key_map, keys);
-  share.base.keystart = share.state.state.key_file_length=
-    MY_ALIGN(info_length, myisam_block_size);
+  aligned_key_start= my_round_up_to_next_power(max_key_block_length ?
+                                               max_key_block_length :
+                                               myisam_block_size);
+
+  share.base.keystart= share.state.state.key_file_length=
+    MY_ALIGN(info_length, aligned_key_start);
   share.base.max_key_block_length=max_key_block_length;
   share.base.max_key_length=ALIGN_SIZE(max_key_length+4);
   share.base.records=ci->max_rows;
diff --git a/storage/myisam/mi_key.c b/storage/myisam/mi_key.c
index f8463a0b6b0..526a733e724 100644
--- a/storage/myisam/mi_key.c
+++ b/storage/myisam/mi_key.c
@@ -127,7 +127,7 @@ uint _mi_make_key(register MI_INFO *info, uint keynr, uchar *key,
     }
     if (keyseg->flag & HA_VAR_LENGTH_PART)
     {
-      uint pack_length= keyseg->bit_start;
+      uint pack_length= (keyseg->bit_start == 1 ? 1 : 2);
       uint tmp_length= (pack_length == 1 ? (uint) *(uchar*) pos :
                         uint2korr(pos));
       pos+= pack_length;			/* Skip VARCHAR length */
diff --git a/storage/myisam/mi_open.c b/storage/myisam/mi_open.c
index 91bf438035f..b61c1af24da 100644
--- a/storage/myisam/mi_open.c
+++ b/storage/myisam/mi_open.c
@@ -295,7 +295,7 @@ MI_INFO *mi_open(const char *name, int mode, uint open_flags)
 			 &share->data_file_name,strlen(data_name)+1,
 			 &share->state.key_root,keys*sizeof(my_off_t),
 			 &share->state.key_del,
-			 (share->state.header.max_block_size*sizeof(my_off_t)),
+			 (share->state.header.max_block_size_index*sizeof(my_off_t)),
 #ifdef THREAD
 			 &share->key_root_lock,sizeof(rw_lock_t)*keys,
 #endif
@@ -310,7 +310,7 @@ MI_INFO *mi_open(const char *name, int mode, uint open_flags)
 	   (char*) key_root, sizeof(my_off_t)*keys);
     memcpy((char*) share->state.key_del,
 	   (char*) key_del, (sizeof(my_off_t) *
-			     share->state.header.max_block_size));
+			     share->state.header.max_block_size_index));
     strmov(share->unique_file_name, name_buff);
     share->unique_name_length= strlen(name_buff);
     strmov(share->index_file_name,  index_name);
@@ -820,7 +820,7 @@ uint mi_state_info_write(File file, MI_STATE_INFO *state, uint pWrite)
   uchar  buff[MI_STATE_INFO_SIZE + MI_STATE_EXTRA_SIZE];
   uchar *ptr=buff;
   uint	i, keys= (uint) state->header.keys,
-	key_blocks=state->header.max_block_size;
+	key_blocks=state->header.max_block_size_index;
   DBUG_ENTER("mi_state_info_write");
 
   memcpy_fixed(ptr,&state->header,sizeof(state->header));
@@ -886,7 +886,7 @@ uchar *mi_state_info_read(uchar *ptr, MI_STATE_INFO *state)
   ptr +=sizeof(state->header);
   keys=(uint) state->header.keys;
   key_parts=mi_uint2korr(state->header.key_parts);
-  key_blocks=state->header.max_block_size;
+  key_blocks=state->header.max_block_size_index;
 
   state->open_count = mi_uint2korr(ptr);	ptr +=2;
   state->changed= (bool) *ptr++;
@@ -1059,7 +1059,7 @@ char *mi_keydef_read(char *ptr, MI_KEYDEF *keydef)
    keydef->keylength	= mi_uint2korr(ptr);	ptr +=2;
    keydef->minlength	= mi_uint2korr(ptr);	ptr +=2;
    keydef->maxlength	= mi_uint2korr(ptr);	ptr +=2;
-   keydef->block_size	= keydef->block_length/MI_MIN_KEY_BLOCK_LENGTH-1;
+   keydef->block_size_index= keydef->block_length/MI_MIN_KEY_BLOCK_LENGTH-1;
    keydef->underflow_block_length=keydef->block_length/3;
    keydef->version	= 0;			/* Not saved */
    keydef->parser       = &ft_default_parser;
diff --git a/storage/myisam/mi_page.c b/storage/myisam/mi_page.c
index 5240c063fba..a5e2b01ed0f 100644
--- a/storage/myisam/mi_page.c
+++ b/storage/myisam/mi_page.c
@@ -112,8 +112,8 @@ int _mi_dispose(register MI_INFO *info, MI_KEYDEF *keyinfo, my_off_t pos,
   DBUG_ENTER("_mi_dispose");
   DBUG_PRINT("enter",("pos: %ld", (long) pos));
 
-  old_link=info->s->state.key_del[keyinfo->block_size];
-  info->s->state.key_del[keyinfo->block_size]=pos;
+  old_link= info->s->state.key_del[keyinfo->block_size_index];
+  info->s->state.key_del[keyinfo->block_size_index]= pos;
   mi_sizestore(buff,old_link);
   info->s->state.changed|= STATE_NOT_SORTED_PAGES;
   DBUG_RETURN(key_cache_write(info->s->key_cache,
@@ -132,7 +132,8 @@ my_off_t _mi_new(register MI_INFO *info, MI_KEYDEF *keyinfo, int level)
   char buff[8];
   DBUG_ENTER("_mi_new");
 
-  if ((pos=info->s->state.key_del[keyinfo->block_size]) == HA_OFFSET_ERROR)
+  if ((pos= info->s->state.key_del[keyinfo->block_size_index]) ==
+      HA_OFFSET_ERROR)
   {
     if (info->state->key_file_length >=
 	info->s->base.max_key_file_length - keyinfo->block_length)
@@ -152,7 +153,7 @@ my_off_t _mi_new(register MI_INFO *info, MI_KEYDEF *keyinfo, int level)
 			(uint) keyinfo->block_length,0))
       pos= HA_OFFSET_ERROR;
     else
-      info->s->state.key_del[keyinfo->block_size]=mi_sizekorr(buff);
+      info->s->state.key_del[keyinfo->block_size_index]= mi_sizekorr(buff);
   }
   info->s->state.changed|= STATE_NOT_SORTED_PAGES;
   DBUG_PRINT("exit",("Pos: %ld",(long) pos));
diff --git a/storage/myisam/mi_test1.c b/storage/myisam/mi_test1.c
index 0e62b074376..0b78ac8a7ff 100644
--- a/storage/myisam/mi_test1.c
+++ b/storage/myisam/mi_test1.c
@@ -95,6 +95,7 @@ static int run_test(const char *filename)
   /* Define a key over the first column */
   keyinfo[0].seg=keyseg;
   keyinfo[0].keysegs=1;
+  keyinfo[0].block_length= 0;                   /* Default block length */
   keyinfo[0].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[0].seg[0].type= key_type;
   keyinfo[0].seg[0].flag= pack_seg;
diff --git a/storage/myisam/mi_test2.c b/storage/myisam/mi_test2.c
index e77a37d853f..357ebb1b9bc 100644
--- a/storage/myisam/mi_test2.c
+++ b/storage/myisam/mi_test2.c
@@ -95,6 +95,7 @@ int main(int argc, char *argv[])
   keyinfo[0].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[0].keysegs=1;
   keyinfo[0].flag = pack_type;
+  keyinfo[0].block_length= 0;                   /* Default block length */
   keyinfo[1].seg= &glob_keyseg[1][0];
   keyinfo[1].seg[0].start=7;
   keyinfo[1].seg[0].length=6;
@@ -111,6 +112,7 @@ int main(int argc, char *argv[])
   keyinfo[1].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[1].keysegs=2;
   keyinfo[1].flag =0;
+  keyinfo[1].block_length= MI_MIN_KEY_BLOCK_LENGTH;  /* Diff blocklength */
   keyinfo[2].seg= &glob_keyseg[2][0];
   keyinfo[2].seg[0].start=12;
   keyinfo[2].seg[0].length=8;
@@ -121,6 +123,7 @@ int main(int argc, char *argv[])
   keyinfo[2].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[2].keysegs=1;
   keyinfo[2].flag =HA_NOSAME;
+  keyinfo[2].block_length= 0;                   /* Default block length */
   keyinfo[3].seg= &glob_keyseg[3][0];
   keyinfo[3].seg[0].start=0;
   keyinfo[3].seg[0].length=reclength-(use_blob ? 8 : 0);
@@ -132,6 +135,7 @@ int main(int argc, char *argv[])
   keyinfo[3].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[3].keysegs=1;
   keyinfo[3].flag = pack_type;
+  keyinfo[3].block_length= 0;                   /* Default block length */
   keyinfo[4].seg= &glob_keyseg[4][0];
   keyinfo[4].seg[0].start=0;
   keyinfo[4].seg[0].length=5;
@@ -143,6 +147,7 @@ int main(int argc, char *argv[])
   keyinfo[4].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[4].keysegs=1;
   keyinfo[4].flag = pack_type;
+  keyinfo[4].block_length= 0;                   /* Default block length */
   keyinfo[5].seg= &glob_keyseg[5][0];
   keyinfo[5].seg[0].start=0;
   keyinfo[5].seg[0].length=4;
@@ -154,6 +159,7 @@ int main(int argc, char *argv[])
   keyinfo[5].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[5].keysegs=1;
   keyinfo[5].flag = pack_type;
+  keyinfo[5].block_length= 0;                   /* Default block length */
 
   recinfo[0].type=pack_fields ? FIELD_SKIP_PRESPACE : 0;
   recinfo[0].length=7;
@@ -813,7 +819,7 @@ end:
     printf("Write records: %d\nUpdate records: %d\nSame-key-read: %d\nDelete records: %d\n", write_count,update,dupp_keys,opt_delete);
     if (rec_pointer_size)
       printf("Record pointer size:  %d\n",rec_pointer_size);
-    printf("myisam_block_size:    %u\n", myisam_block_size);
+    printf("myisam_block_size:    %lu\n", myisam_block_size);
     if (key_cacheing)
     {
       puts("Key cache used");
@@ -914,13 +920,13 @@ static void get_options(int argc, char **argv)
       }
       break;
     case 'e':				/* myisam_block_length */
-      if ((myisam_block_size=atoi(++pos)) < MI_MIN_KEY_BLOCK_LENGTH ||
+      if ((myisam_block_size= atoi(++pos)) < MI_MIN_KEY_BLOCK_LENGTH ||
 	  myisam_block_size > MI_MAX_KEY_BLOCK_LENGTH)
       {
 	fprintf(stderr,"Wrong myisam_block_length\n");
 	exit(1);
       }
-      myisam_block_size=1 << my_bit_log2(myisam_block_size);
+      myisam_block_size= my_round_up_to_next_power(myisam_block_size);
       break;
     case 'E':				/* myisam_block_length */
       if ((key_cache_block_size=atoi(++pos)) < MI_MIN_KEY_BLOCK_LENGTH ||
@@ -929,7 +935,7 @@ static void get_options(int argc, char **argv)
 	fprintf(stderr,"Wrong key_cache_block_size\n");
 	exit(1);
       }
-      key_cache_block_size=1 << my_bit_log2(key_cache_block_size);
+      key_cache_block_size= my_round_up_to_next_power(key_cache_block_size);
       break;
     case 'f':
       if ((first_key=atoi(++pos)) < 0 || first_key >= MYISAM_KEYS)
diff --git a/storage/myisam/mi_test3.c b/storage/myisam/mi_test3.c
index 4e764c6f971..173fbe64cf5 100644
--- a/storage/myisam/mi_test3.c
+++ b/storage/myisam/mi_test3.c
@@ -76,6 +76,7 @@ int main(int argc,char **argv)
   keyinfo[0].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[0].keysegs=1;
   keyinfo[0].flag = (uint8) HA_PACK_KEY;
+  keyinfo[0].block_length= 0;                   /* Default block length */
   keyinfo[1].seg= &keyseg[1][0];
   keyinfo[1].seg[0].start=8;
   keyinfo[1].seg[0].length=4;		/* Long is always 4 in myisam */
@@ -84,6 +85,7 @@ int main(int argc,char **argv)
   keyinfo[1].key_alg=HA_KEY_ALG_BTREE;
   keyinfo[1].keysegs=1;
   keyinfo[1].flag =HA_NOSAME;
+  keyinfo[1].block_length= 0;                   /* Default block length */
 
   recinfo[0].type=0;
   recinfo[0].length=sizeof(record.id);
diff --git a/storage/myisam/myisamdef.h b/storage/myisam/myisamdef.h
index caf6254c321..0b450de9c03 100644
--- a/storage/myisam/myisamdef.h
+++ b/storage/myisam/myisamdef.h
@@ -55,7 +55,7 @@ typedef struct st_mi_state_info
     uchar keys;				/* number of keys in file */
     uchar uniques;			/* number of UNIQUE definitions */
     uchar language;			/* Language for indexes */
-    uchar max_block_size;		/* max keyblock size */
+    uchar max_block_size_index;		/* max keyblock size */
     uchar fulltext_keys;
     uchar not_used;                     /* To align to 8 */
   } header;
@@ -431,7 +431,7 @@ typedef struct st_mi_sort_param
 #define MI_FOUND_WRONG_KEY 32738	/* Impossible value from ha_key_cmp */
 
 #define MI_MAX_KEY_BLOCK_SIZE	(MI_MAX_KEY_BLOCK_LENGTH/MI_MIN_KEY_BLOCK_LENGTH)
-#define MI_BLOCK_SIZE(key_length,data_pointer,key_pointer) (((((key_length)+(data_pointer)+(key_pointer))*4+(key_pointer)+2)/myisam_block_size+1)*myisam_block_size)
+#define MI_BLOCK_SIZE(key_length,data_pointer,key_pointer,block_size) (((((key_length)+(data_pointer)+(key_pointer))*4+(key_pointer)+2)/(block_size)+1)*(block_size))
 #define MI_MAX_KEYPTR_SIZE	5        /* For calculating block lengths */
 #define MI_MIN_KEYBLOCK_LENGTH	50         /* When to split delete blocks */
 
@@ -742,6 +742,8 @@ my_bool check_table_is_closed(const char *name, const char *where);
 int mi_open_datafile(MI_INFO *info, MYISAM_SHARE *share, File file_to_dup);
 int mi_open_keyfile(MYISAM_SHARE *share);
 void mi_setup_functions(register MYISAM_SHARE *share);
+my_bool mi_dynmap_file(MI_INFO *info, my_off_t size);
+void mi_remap_file(MI_INFO *info, my_off_t size);
 
     /* Functions needed by mi_check */
 volatile int *killed_ptr(MI_CHECK *param);
diff --git a/storage/myisam/myisamlog.c b/storage/myisam/myisamlog.c
index de55b86252c..17af4ab34a2 100644
--- a/storage/myisam/myisamlog.c
+++ b/storage/myisam/myisamlog.c
@@ -475,7 +475,7 @@ static int examine_log(my_string file_name, char **table_names)
       {
 	if (!curr_file_info->closed)
 	  files_open--;
-	VOID(tree_delete(&tree, (gptr) curr_file_info, tree.custom_arg));
+        VOID(tree_delete(&tree, (gptr) curr_file_info, 0, tree.custom_arg));
       }
       break;
     case MI_LOG_EXTRA:
diff --git a/storage/myisam/myisampack.c b/storage/myisam/myisampack.c
index e80a3ffacd9..5b3067cb115 100644
--- a/storage/myisam/myisampack.c
+++ b/storage/myisam/myisampack.c
@@ -2726,6 +2726,7 @@ static int compress_isam_file(PACK_MRG_INFO *mrg, HUFF_COUNTS *huff_counts)
 	  break;
 	}
 	case FIELD_LAST:
+        case FIELD_enum_val_count:
 	  abort();				/* Impossible */
 	}
 	start_pos+=count->max_zero_fill;
@@ -2965,7 +2966,7 @@ static int save_state(MI_INFO *isam_file,PACK_MRG_INFO *mrg,my_off_t new_length,
   mi_clear_all_keys_active(share->state.key_map);
   for (key=0 ; key < share->base.keys ; key++)
     share->state.key_root[key]= HA_OFFSET_ERROR;
-  for (key=0 ; key < share->state.header.max_block_size ; key++)
+  for (key=0 ; key < share->state.header.max_block_size_index ; key++)
     share->state.key_del[key]= HA_OFFSET_ERROR;
   isam_file->state->checksum=crc;       /* Save crc here */
   share->changed=1;			/* Force write of header */
diff --git a/storage/ndb/include/kernel/signaldata/TcKeyReq.hpp b/storage/ndb/include/kernel/signaldata/TcKeyReq.hpp
index a37f3811b25..0dbc4094e67 100644
--- a/storage/ndb/include/kernel/signaldata/TcKeyReq.hpp
+++ b/storage/ndb/include/kernel/signaldata/TcKeyReq.hpp
@@ -39,6 +39,7 @@ class TcKeyReq {
   friend class NdbOperation; 
   friend class NdbIndexOperation;
   friend class NdbScanOperation;
+  friend class NdbBlob;
   friend class DbUtil;
 
   /**
diff --git a/storage/ndb/include/ndbapi/NdbBlob.hpp b/storage/ndb/include/ndbapi/NdbBlob.hpp
index 13bbfa59e50..089b70339de 100644
--- a/storage/ndb/include/ndbapi/NdbBlob.hpp
+++ b/storage/ndb/include/ndbapi/NdbBlob.hpp
@@ -326,6 +326,7 @@ private:
   bool isWriteOp();
   bool isDeleteOp();
   bool isScanOp();
+  bool isTakeOverOp();
   // computations
   Uint32 getPartNumber(Uint64 pos);
   Uint32 getPartCount();
diff --git a/storage/ndb/src/ndbapi/NdbBlob.cpp b/storage/ndb/src/ndbapi/NdbBlob.cpp
index 00f71a2f85b..4e3e13b59bf 100644
--- a/storage/ndb/src/ndbapi/NdbBlob.cpp
+++ b/storage/ndb/src/ndbapi/NdbBlob.cpp
@@ -23,6 +23,7 @@
 #include <NdbBlob.hpp>
 #include "NdbBlobImpl.hpp"
 #include <NdbScanOperation.hpp>
+#include <signaldata/TcKeyReq.hpp>
 #include <NdbEventOperationImpl.hpp>
 
 /*
@@ -392,6 +393,13 @@ NdbBlob::isScanOp()
     theNdbOp->theOperationType == NdbOperation::OpenRangeScanRequest;
 }
 
+inline bool
+NdbBlob::isTakeOverOp()
+{
+  return
+    TcKeyReq::getTakeOverScanFlag(theNdbOp->theScanInfo);
+}
+
 // computations (inline)
 
 inline Uint32
@@ -1527,8 +1535,22 @@ NdbBlob::preExecute(NdbTransaction::ExecType anExecType, bool& batch)
     if (isUpdateOp() || isWriteOp() || isDeleteOp()) {
       // add operation before this one to read head+inline
       NdbOperation* tOp = theNdbCon->getNdbOperation(theTable, theNdbOp);
+      /*
+       * If main op is from take over scan lock, the added read is done
+       * as committed read:
+       *
+       * In normal transactional case, the row is locked by us and
+       * committed read returns same as normal read.
+       *
+       * In current TRUNCATE TABLE, the deleting trans is committed in
+       * batches and then restarted with new trans id.  A normal read
+       * would hang on the scan delete lock and then fail.
+       */
+      NdbOperation::LockMode lockMode =
+        ! isTakeOverOp() ?
+          NdbOperation::LM_Read : NdbOperation::LM_CommittedRead;
       if (tOp == NULL ||
-          tOp->readTuple() == -1 ||
+          tOp->readTuple(lockMode) == -1 ||
           setTableKeyValue(tOp) == -1 ||
           getHeadInlineValue(tOp) == -1) {
         setErrorCode(tOp);
diff --git a/storage/ndb/test/ndbapi/testBlobs.cpp b/storage/ndb/test/ndbapi/testBlobs.cpp
index a1b0c89652e..bf9a8b1bce9 100644
--- a/storage/ndb/test/ndbapi/testBlobs.cpp
+++ b/storage/ndb/test/ndbapi/testBlobs.cpp
@@ -44,6 +44,7 @@ struct Opt {
   bool m_dbg;
   bool m_dbgall;
   const char* m_dbug;
+  bool m_fac;
   bool m_full;
   unsigned m_loop;
   unsigned m_parts;
@@ -72,6 +73,7 @@ struct Opt {
     m_dbg(false),
     m_dbgall(false),
     m_dbug(0),
+    m_fac(false),
     m_full(false),
     m_loop(1),
     m_parts(10),
@@ -110,6 +112,7 @@ printusage()
     << "  -dbg        print debug" << endl
     << "  -dbgall     print also NDB API debug (if compiled in)" << endl
     << "  -dbug opt   dbug options" << endl
+    << "  -fac        fetch across commit in scan delete [" << d.m_fac << "]" << endl
     << "  -full       read/write only full blob values" << endl
     << "  -loop N     loop N times 0=forever [" << d.m_loop << "]" << endl
     << "  -parts N    max parts in blob value [" << d.m_parts << "]" << endl
@@ -1255,23 +1258,11 @@ deleteScan(bool idx)
       CHK((ret = g_ops->nextResult(false)) == 0 || ret == 1 || ret == 2);
       if (++n == g_opt.m_batch || ret == 2) {
         DBG("execute batch: n=" << n << " ret=" << ret);
-        switch (0) {
-        case 0: // works normally
+        if (! g_opt.m_fac) {
           CHK(g_con->execute(NoCommit) == 0);
-          CHK(true || g_con->restart() == 0);
-          break;
-        case 1: // nonsense - g_con is invalid for 2nd batch
-          CHK(g_con->execute(Commit) == 0);
-          CHK(true || g_con->restart() == 0);
-          break;
-        case 2: // DBTC sendSignalErrorRefuseLab
-          CHK(g_con->execute(NoCommit) == 0);
-          CHK(g_con->restart() == 0);
-          break;
-        case 3: // 266 time-out
+        } else {
           CHK(g_con->execute(Commit) == 0);
           CHK(g_con->restart() == 0);
-          break;
         }
         n = 0;
       }
@@ -1817,6 +1808,10 @@ NDB_COMMAND(testOdbcDriver, "testBlobs", "testBlobs", "testBlobs", 65535)
 	continue;
       }
     }
+    if (strcmp(arg, "-fac") == 0) {
+      g_opt.m_fac = true;
+      continue;
+    }
     if (strcmp(arg, "-full") == 0) {
       g_opt.m_full = true;
       continue;
diff --git a/storage/ndb/tools/delete_all.cpp b/storage/ndb/tools/delete_all.cpp
index feedded06ad..fcf9b425bd0 100644
--- a/storage/ndb/tools/delete_all.cpp
+++ b/storage/ndb/tools/delete_all.cpp
@@ -23,17 +23,21 @@
 #include <NDBT.hpp>
 
 static int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab,
-                       bool commit_across_open_cursor, int parallelism=240);
+                       bool fetch_across_commit, int parallelism=240);
 
 NDB_STD_OPTS_VARS;
 
 static const char* _dbname = "TEST_DB";
+static my_bool _transactional = false;
 static struct my_option my_long_options[] =
 {
   NDB_STD_OPTS("ndb_desc"),
   { "database", 'd', "Name of database table is in",
     (gptr*) &_dbname, (gptr*) &_dbname, 0,
     GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0 },
+  { "transactional", 't', "Single transaction (may run out of operations)",
+    (gptr*) &_transactional, (gptr*) &_transactional, 0,
+    GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0 },
   { 0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 static void usage()
@@ -84,18 +88,11 @@ int main(int argc, char** argv){
       ndbout << " Table " << argv[i] << " does not exist!" << endl;
       return NDBT_ProgramExit(NDBT_WRONGARGS);
     }
-    // Check if we have any blobs
-    bool commit_across_open_cursor = true;
-    for (int j = 0; j < pTab->getNoOfColumns(); j++) {
-      NdbDictionary::Column::Type t = pTab->getColumn(j)->getType();
-      if (t == NdbDictionary::Column::Blob ||
-          t == NdbDictionary::Column::Text) {
-        commit_across_open_cursor = false;
-        break;
-      }
-    }
-    ndbout << "Deleting all from " << argv[i] << "...";
-    if(clear_table(&MyNdb, pTab, commit_across_open_cursor) == NDBT_FAILED){
+    ndbout << "Deleting all from " << argv[i];
+    if (! _transactional)
+      ndbout << " (non-transactional)";
+    ndbout << " ...";
+    if(clear_table(&MyNdb, pTab, ! _transactional) == NDBT_FAILED){
       res = NDBT_FAILED;
       ndbout << "FAILED" << endl;
     }
@@ -105,7 +102,7 @@ int main(int argc, char** argv){
 
 
 int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab,
-                bool commit_across_open_cursor, int parallelism)
+                bool fetch_across_commit, int parallelism)
 {
   // Scan all records exclusive and delete 
   // them one by one
@@ -136,7 +133,7 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab,
       }
       goto failed;
     }
-    
+
     pOp = pTrans->getNdbScanOperation(pTab->getName());	
     if (pOp == NULL) {
       goto failed;
@@ -167,7 +164,7 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab,
       } while((check = pOp->nextResult(false)) == 0);
       
       if(check != -1){
-        if (commit_across_open_cursor) {
+        if (fetch_across_commit) {
           check = pTrans->execute(NdbTransaction::Commit);   
           pTrans->restart(); // new tx id
         } else {
@@ -198,7 +195,7 @@ int clear_table(Ndb* pNdb, const NdbDictionary::Table* pTab,
       }
       goto failed;
     }
-    if (! commit_across_open_cursor &&
+    if (! fetch_across_commit &&
         pTrans->execute(NdbTransaction::Commit) != 0) {
       err = pTrans->getNdbError();
       goto failed;