Bug#8283 - OPTIMIZE TABLE causes data loss

OPTIMIZE TABLE with myisam_repair_threads > 1 performs a non-quick 
parallel repair. This means that it does not only rebuild all 
indexes, but also the data file.

Non-quick parallel repair works so that there is one thread per 
index. The first of the threads rebuilds also the new data file.

The problem was that all threads shared the read io cache on the
old data file. If there were holes (deleted records) in the table,
the first thread skipped them, writing only contiguous, non-deleted
records to the new data file. Then it built the new index so that
its entries pointed to the correct record positions. But the other
threads didn't know the new record positions, but put the positions
from the old data file into the index.

The new design is so that there is a shared io cache which is filled
by the first thread (the data file writer) with the new contiguous
records and read by the other threads. Now they know the new record
positions.

Another problem was that for the parallel repair of compressed
tables a common bit_buff and rec_buff was used. I changed it so
that thread specific buffers are used for parallel repair.

A similar problem existed for checksum calculation. I made this
multi-thread safe too.

include/my_sys.h

@@ -325,12 +325,18 @@ typedef int (*IO_CACHE_CALLBACK)(struct st_io_cache*);
 #ifdef THREAD
 typedef struct st_io_cache_share
 {
-  /* to sync on reads into buffer */
-  pthread_mutex_t mutex;
-  pthread_cond_t  cond;
-  int             count, total;
-  /* actual IO_CACHE that filled the buffer */
-  struct st_io_cache *active;
+  pthread_mutex_t       mutex;           /* To sync on reads into buffer. */
+  pthread_cond_t        cond;            /* To wait for signals. */
+  pthread_cond_t        cond_writer;     /* For a synchronized writer. */
+  /* Offset in file corresponding to the first byte of buffer. */
+  my_off_t              pos_in_file;
+  /* If a synchronized write cache is the source of the data. */
+  struct st_io_cache    *source_cache;
+  byte                  *buffer;         /* The read buffer. */
+  byte                  *read_end;       /* Behind last valid byte of buffer. */
+  int                   running_threads; /* threads not in lock. */
+  int                   total_threads;   /* threads sharing the cache. */
+  int                   error;           /* Last error. */
 #ifdef NOT_YET_IMPLEMENTED
   /* whether the structure should be free'd */
   my_bool alloced;
@@ -672,8 +678,8 @@ extern void setup_io_cache(IO_CACHE* info);
 extern int _my_b_read(IO_CACHE *info,byte *Buffer,uint Count);
 #ifdef THREAD
 extern int _my_b_read_r(IO_CACHE *info,byte *Buffer,uint Count);
-extern void init_io_cache_share(IO_CACHE *info,
-				IO_CACHE_SHARE *s, uint num_threads);
+extern void init_io_cache_share(IO_CACHE *read_cache, IO_CACHE_SHARE *cshare,
+                                IO_CACHE *write_cache, uint num_threads);
 extern void remove_io_thread(IO_CACHE *info);
 #endif
 extern int _my_b_seq_read(IO_CACHE *info,byte *Buffer,uint Count);

include/myisam.h

@@ -345,7 +345,7 @@ typedef struct st_mi_check_param
   uint testflag, key_cache_block_size;
   uint8 language;
   my_bool using_global_keycache, opt_lock_memory, opt_follow_links;
-  my_bool retry_repair, force_sort, calc_checksum;
+  my_bool retry_repair, force_sort;
   char temp_filename[FN_REFLEN],*isam_file_name;
   MY_TMPDIR *tmpdir;
   int tmpfile_createflag;

myisam/mi_open.c

@@ -201,7 +201,10 @@ MI_INFO *mi_open(const char *name, int mode, uint open_flags)
 	 ((open_flags & HA_OPEN_ABORT_IF_CRASHED) &&
 	  (my_disable_locking && share->state.open_count))))
     {
-      DBUG_PRINT("error",("Table is marked as crashed"));
+      DBUG_PRINT("error",("Table is marked as crashed. open_flags: %u  "
+                          "changed: %u  open_count: %u  !locking: %d",
+                          open_flags, share->state.changed,
+                          share->state.open_count, my_disable_locking));
       my_errno=((share->state.changed & STATE_CRASHED_ON_REPAIR) ?
 		HA_ERR_CRASHED_ON_REPAIR : HA_ERR_CRASHED_ON_USAGE);
       goto err;

myisam/mi_packrec.c

@@ -101,7 +101,8 @@ static uint fill_and_get_bits(MI_BIT_BUFF *bit_buff,uint count);
 static void fill_buffer(MI_BIT_BUFF *bit_buff);
 static uint max_bit(uint value);
 #ifdef HAVE_MMAP
-static uchar *_mi_mempack_get_block_info(MI_INFO *myisam,MI_BLOCK_INFO *info,
+static uchar *_mi_mempack_get_block_info(MI_INFO *myisam, MI_BIT_BUFF *bit_buff,
+                                         MI_BLOCK_INFO *info, byte **rec_buff_p,
 					 uchar *header);
 #endif
 
@@ -428,13 +429,15 @@ int _mi_read_pack_record(MI_INFO *info, my_off_t filepos, byte *buf)
     DBUG_RETURN(-1);			/* _search() didn't find record */
 
   file=info->dfile;
-  if (_mi_pack_get_block_info(info, &block_info, file, filepos))
+  if (_mi_pack_get_block_info(info, &info->bit_buff, &block_info,
+                              &info->rec_buff, file, filepos))
     goto err;
   if (my_read(file,(byte*) info->rec_buff + block_info.offset ,
 	      block_info.rec_len - block_info.offset, MYF(MY_NABP)))
     goto panic;
   info->update|= HA_STATE_AKTIV;
-  DBUG_RETURN(_mi_pack_rec_unpack(info,buf,info->rec_buff,block_info.rec_len));
+  DBUG_RETURN(_mi_pack_rec_unpack(info, &info->bit_buff, buf,
+                                  info->rec_buff, block_info.rec_len));
 panic:
   my_errno=HA_ERR_WRONG_IN_RECORD;
 err:
@@ -443,8 +446,8 @@ err:
 
 
 
-int _mi_pack_rec_unpack(register MI_INFO *info, register byte *to, byte *from,
-			ulong reclength)
+int _mi_pack_rec_unpack(register MI_INFO *info, MI_BIT_BUFF *bit_buff,
+                        register byte *to, byte *from, ulong reclength)
 {
   byte *end_field;
   reg3 MI_COLUMNDEF *end;
@@ -452,18 +455,18 @@ int _mi_pack_rec_unpack(register MI_INFO *info, register byte *to, byte *from,
   MYISAM_SHARE *share=info->s;
   DBUG_ENTER("_mi_pack_rec_unpack");
 
-  init_bit_buffer(&info->bit_buff, (uchar*) from,reclength);
+  init_bit_buffer(bit_buff, (uchar*) from, reclength);
 
   for (current_field=share->rec, end=current_field+share->base.fields ;
        current_field < end ;
        current_field++,to=end_field)
   {
     end_field=to+current_field->length;
-    (*current_field->unpack)(current_field,&info->bit_buff,(uchar*) to,
+    (*current_field->unpack)(current_field, bit_buff, (uchar*) to,
 			     (uchar*) end_field);
   }
-  if (! info->bit_buff.error &&
-      info->bit_buff.pos - info->bit_buff.bits/8 == info->bit_buff.end)
+  if (!bit_buff->error &&
+      bit_buff->pos - bit_buff->bits / 8 == bit_buff->end)
     DBUG_RETURN(0);
   info->update&= ~HA_STATE_AKTIV;
   DBUG_RETURN(my_errno=HA_ERR_WRONG_IN_RECORD);
@@ -977,13 +980,16 @@ int _mi_read_rnd_pack_record(MI_INFO *info, byte *buf,
 
   if (info->opt_flag & READ_CACHE_USED)
   {
-    if (_mi_read_cache(&info->rec_cache,(byte*) block_info.header,filepos,
-		      share->pack.ref_length, skip_deleted_blocks))
+    if (_mi_read_cache(&info->rec_cache, (byte*) block_info.header,
+                       filepos, share->pack.ref_length,
+                       skip_deleted_blocks ? READING_NEXT : 0))
       goto err;
-    b_type=_mi_pack_get_block_info(info,&block_info,-1, filepos);
+    b_type=_mi_pack_get_block_info(info, &info->bit_buff, &block_info,
+                                   &info->rec_buff, -1, filepos);
   }
   else
-    b_type=_mi_pack_get_block_info(info,&block_info,info->dfile,filepos);
+    b_type=_mi_pack_get_block_info(info, &info->bit_buff, &block_info,
+                                   &info->rec_buff, info->dfile, filepos);
   if (b_type)
     goto err;					/* Error code is already set */
 #ifndef DBUG_OFF
@@ -996,9 +1002,9 @@ int _mi_read_rnd_pack_record(MI_INFO *info, byte *buf,
 
   if (info->opt_flag & READ_CACHE_USED)
   {
-    if (_mi_read_cache(&info->rec_cache,(byte*) info->rec_buff,
+    if (_mi_read_cache(&info->rec_cache, (byte*) info->rec_buff,
                        block_info.filepos, block_info.rec_len,
-		      skip_deleted_blocks))
+                       skip_deleted_blocks ? READING_NEXT : 0))
       goto err;
   }
   else
@@ -1013,8 +1019,8 @@ int _mi_read_rnd_pack_record(MI_INFO *info, byte *buf,
   info->nextpos=block_info.filepos+block_info.rec_len;
   info->update|= HA_STATE_AKTIV | HA_STATE_KEY_CHANGED;
 
-  DBUG_RETURN (_mi_pack_rec_unpack(info,buf,info->rec_buff,
-				   block_info.rec_len));
+  DBUG_RETURN (_mi_pack_rec_unpack(info, &info->bit_buff, buf,
+                                   info->rec_buff, block_info.rec_len));
  err:
   DBUG_RETURN(my_errno);
 }
@@ -1022,8 +1028,9 @@ int _mi_read_rnd_pack_record(MI_INFO *info, byte *buf,
 
 	/* Read and process header from a huff-record-file */
 
-uint _mi_pack_get_block_info(MI_INFO *myisam, MI_BLOCK_INFO *info, File file,
-			     my_off_t filepos)
+uint _mi_pack_get_block_info(MI_INFO *myisam, MI_BIT_BUFF *bit_buff,
+                             MI_BLOCK_INFO *info, byte **rec_buff_p,
+                             File file, my_off_t filepos)
 {
   uchar *header=info->header;
   uint head_length,ref_length;
@@ -1048,17 +1055,17 @@ uint _mi_pack_get_block_info(MI_INFO *myisam, MI_BLOCK_INFO *info, File file,
     head_length+= read_pack_length((uint) myisam->s->pack.version,
                                    header + head_length, &info->blob_len);
     if (!(mi_alloc_rec_buff(myisam,info->rec_len + info->blob_len,
-			    &myisam->rec_buff)))
+			    rec_buff_p)))
       return BLOCK_FATAL_ERROR;			/* not enough memory */
-    myisam->bit_buff.blob_pos=(uchar*) myisam->rec_buff+info->rec_len;
-    myisam->bit_buff.blob_end= myisam->bit_buff.blob_pos+info->blob_len;
+    bit_buff->blob_pos= (uchar*) *rec_buff_p + info->rec_len;
+    bit_buff->blob_end= bit_buff->blob_pos + info->blob_len;
     myisam->blob_length=info->blob_len;
   }
   info->filepos=filepos+head_length;
   if (file > 0)
   {
     info->offset=min(info->rec_len, ref_length - head_length);
-    memcpy(myisam->rec_buff, header+head_length, info->offset);
+    memcpy(*rec_buff_p, header + head_length, info->offset);
   }
   return 0;
 }
@@ -1198,7 +1205,8 @@ void _mi_unmap_file(MI_INFO *info)
 }
 
 
-static uchar *_mi_mempack_get_block_info(MI_INFO *myisam,MI_BLOCK_INFO *info,
+static uchar *_mi_mempack_get_block_info(MI_INFO *myisam, MI_BIT_BUFF *bit_buff,
+                                         MI_BLOCK_INFO *info, byte **rec_buff_p,
 					 uchar *header)
 {
   header+= read_pack_length((uint) myisam->s->pack.version, header,
@@ -1209,10 +1217,10 @@ static uchar *_mi_mempack_get_block_info(MI_INFO *myisam,MI_BLOCK_INFO *info,
                               &info->blob_len);
     /* mi_alloc_rec_buff sets my_errno on error */
     if (!(mi_alloc_rec_buff(myisam, info->blob_len,
-			    &myisam->rec_buff)))
+			    rec_buff_p)))
       return 0;				/* not enough memory */
-    myisam->bit_buff.blob_pos=(uchar*) myisam->rec_buff;
-    myisam->bit_buff.blob_end= (uchar*) myisam->rec_buff + info->blob_len;
+    bit_buff->blob_pos= (uchar*) *rec_buff_p;
+    bit_buff->blob_end= (uchar*) *rec_buff_p + info->blob_len;
   }
   return header;
 }
@@ -1228,11 +1236,13 @@ static int _mi_read_mempack_record(MI_INFO *info, my_off_t filepos, byte *buf)
   if (filepos == HA_OFFSET_ERROR)
     DBUG_RETURN(-1);			/* _search() didn't find record */
 
-  if (!(pos= (byte*) _mi_mempack_get_block_info(info,&block_info,
+  if (!(pos= (byte*) _mi_mempack_get_block_info(info, &info->bit_buff,
+                                                &block_info, &info->rec_buff,
 						(uchar*) share->file_map+
 						filepos)))
     DBUG_RETURN(-1);
-  DBUG_RETURN(_mi_pack_rec_unpack(info, buf, pos, block_info.rec_len));
+  DBUG_RETURN(_mi_pack_rec_unpack(info, &info->bit_buff, buf,
+                                  pos, block_info.rec_len));
 }
 
 
@@ -1252,7 +1262,8 @@ static int _mi_read_rnd_mempack_record(MI_INFO *info, byte *buf,
     my_errno=HA_ERR_END_OF_FILE;
     goto err;
   }
-  if (!(pos= (byte*) _mi_mempack_get_block_info(info,&block_info,
+  if (!(pos= (byte*) _mi_mempack_get_block_info(info, &info->bit_buff,
+                                                &block_info, &info->rec_buff,
 						(uchar*)
 						(start=share->file_map+
 						 filepos))))
@@ -1269,7 +1280,8 @@ static int _mi_read_rnd_mempack_record(MI_INFO *info, byte *buf,
   info->nextpos=filepos+(uint) (pos-start)+block_info.rec_len;
   info->update|= HA_STATE_AKTIV | HA_STATE_KEY_CHANGED;
 
-  DBUG_RETURN (_mi_pack_rec_unpack(info,buf,pos, block_info.rec_len));
+  DBUG_RETURN (_mi_pack_rec_unpack(info, &info->bit_buff, buf,
+                                   pos, block_info.rec_len));
  err:
   DBUG_RETURN(my_errno);
 }

myisam/myisamdef.h

@@ -75,7 +75,7 @@ typedef struct st_mi_state_info
   ulong sec_index_changed;		/* Updated when new sec_index */
   ulong sec_index_used;			/* which extra index are in use */
   ulonglong key_map;			/* Which keys are in use */
-  ha_checksum checksum;
+  ha_checksum checksum;                 /* Table checksum */
   ulong version;			/* timestamp of create */
   time_t create_time;			/* Time when created database */
   time_t recover_time;			/* Time for last recover */
@@ -176,6 +176,7 @@ typedef struct st_mi_isam_share {	/* Shared between opens */
   int (*delete_record)(struct st_myisam_info*);
   int (*read_rnd)(struct st_myisam_info*, byte*, my_off_t, my_bool);
   int (*compare_record)(struct st_myisam_info*, const byte *);
+  /* Function to use for a row checksum. */
   ha_checksum (*calc_checksum)(struct st_myisam_info*, const byte *);
   int (*compare_unique)(struct st_myisam_info*, MI_UNIQUEDEF *,
 			const byte *record, my_off_t pos);
@@ -249,7 +250,7 @@ struct st_myisam_info {
   my_off_t last_keypage;		/* Last key page read */
   my_off_t last_search_keypage;		/* Last keypage when searching */
   my_off_t dupp_key_pos;
-  ha_checksum checksum;
+  ha_checksum checksum;                 /* Temp storage for row checksum */
   /* QQ: the folloing two xxx_length fields should be removed,
      as they are not compatible with parallel repair */
   ulong packed_length,blob_length;	/* Length of found, packed record */
@@ -297,6 +298,7 @@ typedef struct st_mi_sort_param
   pthread_t  thr;
   IO_CACHE read_cache, tempfile, tempfile_for_exceptions;
   DYNAMIC_ARRAY buffpek;
+  MI_BIT_BUFF   bit_buff;               /* For parallel repair of packrec. */
 
   /*
     The next two are used to collect statistics, see update_key_parts for
@@ -309,6 +311,7 @@ typedef struct st_mi_sort_param
   uint key, key_length,real_key_length,sortbuff_size;
   uint maxbuffers, keys, find_length, sort_keys_length;
   my_bool fix_datafile, master;
+  my_bool calc_checksum;                /* calculate table checksum */
   MI_KEYDEF *keyinfo;
   HA_KEYSEG *seg;
   SORT_INFO *sort_info;
@@ -361,8 +364,15 @@ typedef struct st_mi_sort_param
 #define mi_putint(x,y,nod) { uint16 boh=(nod ? (uint16) 32768 : 0) + (uint16) (y);\
 			  mi_int2store(x,boh); }
 #define mi_test_if_nod(x) (x[0] & 128 ? info->s->base.key_reflength : 0)
-#define mi_mark_crashed(x) (x)->s->state.changed|=STATE_CRASHED
-#define mi_mark_crashed_on_repair(x) { (x)->s->state.changed|=STATE_CRASHED|STATE_CRASHED_ON_REPAIR ; (x)->update|= HA_STATE_CHANGED; }
+#define mi_mark_crashed(x) do{(x)->s->state.changed|= STATE_CRASHED; \
+                              DBUG_PRINT("error", ("Marked table crashed")); \
+                           }while(0)
+#define mi_mark_crashed_on_repair(x) do{(x)->s->state.changed|= \
+                                        STATE_CRASHED|STATE_CRASHED_ON_REPAIR; \
+                                        (x)->update|= HA_STATE_CHANGED; \
+                                        DBUG_PRINT("error", \
+                                                   ("Marked table crashed")); \
+                                     }while(0)
 #define mi_is_crashed(x) ((x)->s->state.changed & STATE_CRASHED)
 #define mi_is_crashed_on_repair(x) ((x)->s->state.changed & STATE_CRASHED_ON_REPAIR)
 
@@ -600,8 +610,8 @@ extern void _mi_print_key(FILE *stream,HA_KEYSEG *keyseg,const uchar *key,
 extern my_bool _mi_read_pack_info(MI_INFO *info,pbool fix_keys);
 extern int _mi_read_pack_record(MI_INFO *info,my_off_t filepos,byte *buf);
 extern int _mi_read_rnd_pack_record(MI_INFO*, byte *,my_off_t, my_bool);
-extern int _mi_pack_rec_unpack(MI_INFO *info,byte *to,byte *from,
-			       ulong reclength);
+extern int _mi_pack_rec_unpack(MI_INFO *info, MI_BIT_BUFF *bit_buff,
+                               byte *to, byte *from, ulong reclength);
 extern ulonglong mi_safe_mul(ulonglong a,ulonglong b);
 extern int _mi_ft_update(MI_INFO *info, uint keynr, byte *keybuf,
 			 const byte *oldrec, const byte *newrec, my_off_t pos);
@@ -666,7 +676,9 @@ extern "C" {
 
 extern uint _mi_get_block_info(MI_BLOCK_INFO *,File, my_off_t);
 extern uint _mi_rec_pack(MI_INFO *info,byte *to,const byte *from);
-extern uint _mi_pack_get_block_info(MI_INFO *, MI_BLOCK_INFO *, File, my_off_t);
+extern uint _mi_pack_get_block_info(MI_INFO *myisam, MI_BIT_BUFF *bit_buff,
+                                    MI_BLOCK_INFO *info, byte **rec_buff_p,
+                                    File file, my_off_t filepos);
 extern void _my_store_blob_length(byte *pos,uint pack_length,uint length);
 extern void _myisam_log(enum myisam_log_commands command,MI_INFO *info,
 		       const byte *buffert,uint length);

myisam/sort.c

@@ -309,7 +309,7 @@ static ha_rows NEAR_F find_all_keys(MI_SORT_PARAM *info, uint keys,
 
 pthread_handler_decl(thr_find_all_keys,arg)
 {
-  MI_SORT_PARAM *info= (MI_SORT_PARAM*) arg;
+  MI_SORT_PARAM *sort_param= (MI_SORT_PARAM*) arg;
   int error;
   uint memavl,old_memavl,keys,sort_length;
   uint idx, maxbuffer;
@@ -321,31 +321,33 @@ pthread_handler_decl(thr_find_all_keys,arg)
 
   if (my_thread_init())
     goto err;
-  if (info->sort_info->got_error)
+  DBUG_ENTER("thr_find_all_keys");
+  DBUG_PRINT("enter", ("master: %d", sort_param->master));
+  if (sort_param->sort_info->got_error)
     goto err;
 
-  if (info->keyinfo->flag && HA_VAR_LENGTH_KEY)
+  if (sort_param->keyinfo->flag && HA_VAR_LENGTH_KEY)
   {
-    info->write_keys=write_keys_varlen;
-    info->read_to_buffer=read_to_buffer_varlen;
-    info->write_key=write_merge_key_varlen;
+    sort_param->write_keys=     write_keys_varlen;
+    sort_param->read_to_buffer= read_to_buffer_varlen;
+    sort_param->write_key=      write_merge_key_varlen;
   }
   else
   {
-    info->write_keys=write_keys;
-    info->read_to_buffer=read_to_buffer;
-    info->write_key=write_merge_key;
+    sort_param->write_keys=     write_keys;
+    sort_param->read_to_buffer= read_to_buffer;
+    sort_param->write_key=      write_merge_key;
   }
 
-  my_b_clear(&info->tempfile);
-  my_b_clear(&info->tempfile_for_exceptions);
-  bzero((char*) &info->buffpek,sizeof(info->buffpek));
-  bzero((char*) &info->unique, sizeof(info->unique));
+  my_b_clear(&sort_param->tempfile);
+  my_b_clear(&sort_param->tempfile_for_exceptions);
+  bzero((char*) &sort_param->buffpek, sizeof(sort_param->buffpek));
+  bzero((char*) &sort_param->unique,  sizeof(sort_param->unique));
   sort_keys= (uchar **) NULL;
 
-  memavl=max(info->sortbuff_size, MIN_SORT_MEMORY);
-  idx=      info->sort_info->max_records;
-  sort_length=  info->key_length;
+  memavl=       max(sort_param->sortbuff_size, MIN_SORT_MEMORY);
+  idx=          sort_param->sort_info->max_records;
+  sort_length=  sort_param->key_length;
   maxbuffer=    1;
 
   while (memavl >= MIN_SORT_MEMORY)
@@ -363,18 +365,19 @@ pthread_handler_decl(thr_find_all_keys,arg)
             (keys=(memavl-sizeof(BUFFPEK)*maxbuffer)/
              (sort_length+sizeof(char*))) <= 1)
         {
-          mi_check_print_error(info->sort_info->param,
+          mi_check_print_error(sort_param->sort_info->param,
                                "sort_buffer_size is to small");
           goto err;
         }
       }
       while ((maxbuffer= (int) (idx/(keys-1)+1)) != skr);
     }
-    if ((sort_keys=(uchar **)my_malloc(keys*(sort_length+sizeof(char*))+
-				       ((info->keyinfo->flag & HA_FULLTEXT) ?
+    if ((sort_keys= (uchar**)
+         my_malloc(keys*(sort_length+sizeof(char*))+
+                   ((sort_param->keyinfo->flag & HA_FULLTEXT) ?
                     HA_FT_MAXBYTELEN : 0), MYF(0))))
     {
-      if (my_init_dynamic_array(&info->buffpek, sizeof(BUFFPEK),
+      if (my_init_dynamic_array(&sort_param->buffpek, sizeof(BUFFPEK),
 				maxbuffer, maxbuffer/2))
         my_free((gptr) sort_keys,MYF(0));
       else
@@ -386,69 +389,87 @@ pthread_handler_decl(thr_find_all_keys,arg)
   }
   if (memavl < MIN_SORT_MEMORY)
   {
-    mi_check_print_error(info->sort_info->param,"Sort buffer to small"); /* purecov: tested */
+    mi_check_print_error(sort_param->sort_info->param, "Sort buffer too small");
     goto err; /* purecov: tested */
   }
 
-  if (info->sort_info->param->testflag & T_VERBOSE)
-    printf("Key %d - Allocating buffer for %d keys\n",info->key+1,keys);
-  info->sort_keys=sort_keys;
+  if (sort_param->sort_info->param->testflag & T_VERBOSE)
+    printf("Key %d - Allocating buffer for %d keys\n",
+           sort_param->key + 1, keys);
+  sort_param->sort_keys= sort_keys;
 
   idx=error=0;
   sort_keys[0]=(uchar*) (sort_keys+keys);
 
-  while (!(error=info->sort_info->got_error) &&
-         !(error=(*info->key_read)(info,sort_keys[idx])))
+  DBUG_PRINT("info", ("reading keys"));
+  while (!(error= sort_param->sort_info->got_error) &&
+         !(error= (*sort_param->key_read)(sort_param, sort_keys[idx])))
   {
-    if (info->real_key_length > info->key_length)
+    if (sort_param->real_key_length > sort_param->key_length)
     {
-      if (write_key(info,sort_keys[idx], &info->tempfile_for_exceptions))
+      if (write_key(sort_param, sort_keys[idx],
+                    &sort_param->tempfile_for_exceptions))
         goto err;
       continue;
     }
 
     if (++idx == keys)
     {
-      if (info->write_keys(info,sort_keys,idx-1,
-		     (BUFFPEK *)alloc_dynamic(&info->buffpek),
-		     &info->tempfile))
+      if (sort_param->write_keys(sort_param, sort_keys, idx - 1,
+                                 (BUFFPEK*) alloc_dynamic(&sort_param->buffpek),
+                                 &sort_param->tempfile))
         goto err;
       sort_keys[0]=(uchar*) (sort_keys+keys);
-      memcpy(sort_keys[0],sort_keys[idx-1],(size_t) info->key_length);
+      memcpy(sort_keys[0], sort_keys[idx - 1], (size_t) sort_param->key_length);
       idx=1;
     }
-    sort_keys[idx]=sort_keys[idx-1]+info->key_length;
+    sort_keys[idx]= sort_keys[idx - 1] + sort_param->key_length;
   }
   if (error > 0)
     goto err;
-  if (info->buffpek.elements)
+  if (sort_param->buffpek.elements)
   {
-    if (info->write_keys(info,sort_keys, idx,
-		   (BUFFPEK *) alloc_dynamic(&info->buffpek), &info->tempfile))
+    if (sort_param->write_keys(sort_param, sort_keys, idx,
+                               (BUFFPEK*) alloc_dynamic(&sort_param->buffpek),
+                               &sort_param->tempfile))
       goto err;
-    info->keys=(info->buffpek.elements-1)*(keys-1)+idx;
+    sort_param->keys= (sort_param->buffpek.elements - 1) * (keys - 1) + idx;
   }
   else
-    info->keys=idx;
+    sort_param->keys= idx;
 
-  info->sort_keys_length=keys;
+  sort_param->sort_keys_length= keys;
   goto ok;
 
 err:
-  info->sort_info->got_error=1; /* no need to protect this with a mutex */
+  DBUG_PRINT("error", ("got some error"));
+  sort_param->sort_info->got_error= 1; /* no need to protect with a mutex */
   if (sort_keys)
     my_free((gptr) sort_keys,MYF(0));
-  info->sort_keys=0;
-  delete_dynamic(& info->buffpek);
-  close_cached_file(&info->tempfile);
-  close_cached_file(&info->tempfile_for_exceptions);
+  sort_param->sort_keys= 0;
+  delete_dynamic(& sort_param->buffpek);
+  close_cached_file(&sort_param->tempfile);
+  close_cached_file(&sort_param->tempfile_for_exceptions);
 
 ok:
-  remove_io_thread(&info->read_cache);
-  pthread_mutex_lock(&info->sort_info->mutex);
-  info->sort_info->threads_running--;
-  pthread_cond_signal(&info->sort_info->cond);
-  pthread_mutex_unlock(&info->sort_info->mutex);
+  /*
+    Detach from the share if the writer is involved. Avoid others to
+    be blocked. This includes a flush of the write buffer. This will
+    also indicate EOF to the readers.
+  */
+  if (sort_param->sort_info->info->rec_cache.share)
+    remove_io_thread(&sort_param->sort_info->info->rec_cache);
+
+  /* Readers detach from the share if any. Avoid others to be blocked. */
+  if (sort_param->read_cache.share)
+    remove_io_thread(&sort_param->read_cache);
+
+  pthread_mutex_lock(&sort_param->sort_info->mutex);
+  if (!--sort_param->sort_info->threads_running)
+    pthread_cond_signal(&sort_param->sort_info->cond);
+  pthread_mutex_unlock(&sort_param->sort_info->mutex);
+
+  DBUG_PRINT("exit", ("======== ending thread ========"));
   my_thread_end();
   return NULL;
 }
@@ -466,6 +487,7 @@ int thr_write_keys(MI_SORT_PARAM *sort_param)
   MYISAM_SHARE *share=info->s;
   MI_SORT_PARAM *sinfo;
   byte *mergebuf=0;
+  DBUG_ENTER("thr_write_keys");
   LINT_INIT(length);
 
   for (i= 0, sinfo= sort_param ;
@@ -604,7 +626,7 @@ int thr_write_keys(MI_SORT_PARAM *sort_param)
     }
   }
   my_free((gptr) mergebuf,MYF(MY_ALLOW_ZERO_PTR));
-  return got_error;
+  DBUG_RETURN(got_error);
 }
 #endif /* THREAD */
 

mysql-test/r/myisam.result

@@ -756,3 +756,129 @@ a	b
 xxxxxxxxx	bbbbbb
 xxxxxxxxx	bbbbbb
 DROP TABLE t1;
+SET @@myisam_repair_threads=2;
+SHOW VARIABLES LIKE 'myisam_repair%';
+Variable_name	Value
+myisam_repair_threads	2
+CREATE TABLE t1 (
+`_id` int(11) NOT NULL default '0',
+`url` text,
+`email` text,
+`description` text,
+`loverlap` int(11) default NULL,
+`roverlap` int(11) default NULL,
+`lneighbor_id` int(11) default NULL,
+`rneighbor_id` int(11) default NULL,
+`length_` int(11) default NULL,
+`sequence` mediumtext,
+`name` text,
+`_obj_class` text NOT NULL,
+PRIMARY KEY  (`_id`),
+UNIQUE KEY `sequence_name_index` (`name`(50)),
+KEY (`length_`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1;
+INSERT INTO t1 VALUES
+(1,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample1',''),
+(2,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample2',''),
+(3,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample3',''),
+(4,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample4',''),
+(5,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample5',''),
+(6,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample6',''),
+(7,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample7',''),
+(8,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample8',''),
+(9,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample9','');
+SELECT _id FROM t1;
+_id
+1
+2
+3
+4
+5
+6
+7
+8
+9
+DELETE FROM t1 WHERE _id < 8;
+SHOW TABLE STATUS LIKE 't1';
+Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
+t1	MyISAM	9	Dynamic	2	#	#	#	#	140	#	#	#	#	#	#		
+CHECK TABLE t1 EXTENDED;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+OPTIMIZE TABLE t1;
+Table	Op	Msg_type	Msg_text
+test.t1	optimize	status	OK
+CHECK TABLE t1 EXTENDED;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SHOW TABLE STATUS LIKE 't1';
+Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
+t1	MyISAM	9	Dynamic	2	#	#	#	#	0	#	#	#	#	#	#		
+SELECT _id FROM t1;
+_id
+8
+9
+DROP TABLE t1;
+CREATE TABLE t1 (
+`_id` int(11) NOT NULL default '0',
+`url` text,
+`email` text,
+`description` text,
+`loverlap` int(11) default NULL,
+`roverlap` int(11) default NULL,
+`lneighbor_id` int(11) default NULL,
+`rneighbor_id` int(11) default NULL,
+`length_` int(11) default NULL,
+`sequence` mediumtext,
+`name` text,
+`_obj_class` text NOT NULL,
+PRIMARY KEY  (`_id`),
+UNIQUE KEY `sequence_name_index` (`name`(50)),
+KEY (`length_`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1;
+INSERT INTO t1 VALUES
+(1,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample1',''),
+(2,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample2',''),
+(3,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample3',''),
+(4,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample4',''),
+(5,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample5',''),
+(6,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample6',''),
+(7,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample7',''),
+(8,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample8',''),
+(9,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample9','');
+SELECT _id FROM t1;
+_id
+1
+2
+3
+4
+5
+6
+7
+8
+9
+DELETE FROM t1 WHERE _id < 8;
+SHOW TABLE STATUS LIKE 't1';
+Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
+t1	MyISAM	9	Dynamic	2	#	#	#	#	140	#	#	#	#	#	#		
+CHECK TABLE t1 EXTENDED;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+REPAIR TABLE t1 QUICK;
+Table	Op	Msg_type	Msg_text
+test.t1	repair	status	OK
+CHECK TABLE t1 EXTENDED;
+Table	Op	Msg_type	Msg_text
+test.t1	check	status	OK
+SHOW TABLE STATUS LIKE 't1';
+Name	Engine	Version	Row_format	Rows	Avg_row_length	Data_length	Max_data_length	Index_length	Data_free	Auto_increment	Create_time	Update_time	Check_time	Collation	Checksum	Create_options	Comment
+t1	MyISAM	9	Dynamic	2	#	#	#	#	140	#	#	#	#	#	#		
+SELECT _id FROM t1;
+_id
+8
+9
+DROP TABLE t1;
+SET @@myisam_repair_threads=1;
+SHOW VARIABLES LIKE 'myisam_repair%';
+Variable_name	Value
+myisam_repair_threads	1

mysql-test/t/myisam.test

@@ -714,4 +714,97 @@ UPDATE t1 AS ta1,t1 AS ta2 SET ta1.b='aaaaaa',ta2.b='bbbbbb';
 SELECT * FROM t1;
 DROP TABLE t1;
 
+#
+# Bug#8283 - OPTIMIZE TABLE causes data loss
+#
+SET @@myisam_repair_threads=2;
+SHOW VARIABLES LIKE 'myisam_repair%';
+#
+# Test OPTIMIZE. This creates a new data file.
+CREATE TABLE t1 (
+  `_id` int(11) NOT NULL default '0',
+  `url` text,
+  `email` text,
+  `description` text,
+  `loverlap` int(11) default NULL,
+  `roverlap` int(11) default NULL,
+  `lneighbor_id` int(11) default NULL,
+  `rneighbor_id` int(11) default NULL,
+  `length_` int(11) default NULL,
+  `sequence` mediumtext,
+  `name` text,
+  `_obj_class` text NOT NULL,
+  PRIMARY KEY  (`_id`),
+  UNIQUE KEY `sequence_name_index` (`name`(50)),
+  KEY (`length_`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1;
+#
+INSERT INTO t1 VALUES
+  (1,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample1',''),
+  (2,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample2',''),
+  (3,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample3',''),
+  (4,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample4',''),
+  (5,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample5',''),
+  (6,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample6',''),
+  (7,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample7',''),
+  (8,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample8',''),
+  (9,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample9','');
+#
+SELECT _id FROM t1;
+DELETE FROM t1 WHERE _id < 8;
+--replace_column 6 # 7 # 8 # 9 # 11 # 12 # 13 # 14 # 15 # 16 #
+SHOW TABLE STATUS LIKE 't1';
+CHECK TABLE t1 EXTENDED;
+OPTIMIZE TABLE t1;
+CHECK TABLE t1 EXTENDED;
+--replace_column 6 # 7 # 8 # 9 # 11 # 12 # 13 # 14 # 15 # 16 #
+SHOW TABLE STATUS LIKE 't1';
+SELECT _id FROM t1;
+DROP TABLE t1;
+#
+# Test REPAIR QUICK. This retains the old data file.
+CREATE TABLE t1 (
+  `_id` int(11) NOT NULL default '0',
+  `url` text,
+  `email` text,
+  `description` text,
+  `loverlap` int(11) default NULL,
+  `roverlap` int(11) default NULL,
+  `lneighbor_id` int(11) default NULL,
+  `rneighbor_id` int(11) default NULL,
+  `length_` int(11) default NULL,
+  `sequence` mediumtext,
+  `name` text,
+  `_obj_class` text NOT NULL,
+  PRIMARY KEY  (`_id`),
+  UNIQUE KEY `sequence_name_index` (`name`(50)),
+  KEY (`length_`)
+) ENGINE=MyISAM DEFAULT CHARSET=latin1;
+#
+INSERT INTO t1 VALUES
+  (1,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample1',''),
+  (2,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample2',''),
+  (3,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample3',''),
+  (4,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample4',''),
+  (5,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample5',''),
+  (6,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample6',''),
+  (7,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample7',''),
+  (8,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample8',''),
+  (9,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,NULL,'sample9','');
+#
+SELECT _id FROM t1;
+DELETE FROM t1 WHERE _id < 8;
+--replace_column 6 # 7 # 8 # 9 # 11 # 12 # 13 # 14 # 15 # 16 #
+SHOW TABLE STATUS LIKE 't1';
+CHECK TABLE t1 EXTENDED;
+REPAIR TABLE t1 QUICK;
+CHECK TABLE t1 EXTENDED;
+--replace_column 6 # 7 # 8 # 9 # 11 # 12 # 13 # 14 # 15 # 16 #
+SHOW TABLE STATUS LIKE 't1';
+SELECT _id FROM t1;
+DROP TABLE t1;
+#
+SET @@myisam_repair_threads=1;
+SHOW VARIABLES LIKE 'myisam_repair%';
+
 # End of 4.1 tests