merge fix

46 files changed, 1285 insertions, 647 deletions

diff --git a/sql/filesort.cc b/sql/filesort.cc
index 4c2ba1e1a59..928138b8d48 100644
--- a/sql/filesort.cc
+++ b/sql/filesort.cc
@@ -1,4 +1,5 @@
-/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
+
+/* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB  
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -51,6 +52,10 @@ static int merge_index(SORTPARAM *param,uchar *sort_buffer,
 static bool save_index(SORTPARAM *param,uchar **sort_keys, uint count);
 static uint sortlength(SORT_FIELD *sortorder, uint s_length,
 		       bool *multi_byte_charset);
+static SORT_ADDON_FIELD *get_addon_fields(THD *thd, Field **ptabfield,
+                                          uint sortlength, uint *plength);
+static void unpack_addon_fields(struct st_sort_addon_field *addon_field,
+                                byte *buff);
 
 	/*
 	  Creates a set of pointers that can be used to read the rows
@@ -82,16 +87,48 @@ ha_rows filesort(THD *thd, TABLE *table, SORT_FIELD *sortorder, uint s_length,
   DBUG_PUSH("");		/* No DBUG here */
 #endif
 
-  outfile= table->io_cache;
+  outfile= table->sort.io_cache;
   my_b_clear(&tempfile);
   my_b_clear(&buffpek_pointers);
   buffpek=0;
   sort_keys= (uchar **) NULL;
   error= 1;
   bzero((char*) &param,sizeof(param));
+  param.sort_length= sortlength(sortorder, s_length, &multi_byte_charset);
   param.ref_length= table->file->ref_length;
-  param.sort_length= (sortlength(sortorder,s_length, &multi_byte_charset)+
-		      param.ref_length);
+  param.addon_field= 0;
+  param.addon_length= 0;
+  if (!(table->tmp_table || table->fulltext_searched))
+  {
+    /* 
+      Get the descriptors of all fields whose values are appended 
+      to sorted fields and get its total length in param.spack_length.
+    */
+    param.addon_field= get_addon_fields(thd, table->field, 
+                                        param.sort_length,
+                                        &param.addon_length);
+  }
+  table->sort.addon_buf= 0;
+  table->sort.addon_length= param.addon_length;
+  table->sort.addon_field= param.addon_field;
+  table->sort.unpack= unpack_addon_fields;
+  if (param.addon_field)
+  {
+    param.res_length= param.addon_length;
+    if (!(table->sort.addon_buf= (byte *) my_malloc(param.addon_length,
+                                                    MYF(MY_WME))))
+      goto err;
+  }
+  else
+  {
+    param.res_length= param.ref_length;
+    /* 
+      The reference to the record is considered 
+      as an additional sorted field
+    */
+    param.sort_length+= param.ref_length;
+  }
+  param.rec_length= param.sort_length+param.addon_length;
   param.max_rows= max_rows;
 
   if (select && select->quick)
@@ -115,7 +152,7 @@ ha_rows filesort(THD *thd, TABLE *table, SORT_FIELD *sortorder, uint s_length,
     records=table->file->estimate_number_of_rows();
     selected_records_file= 0;
   }
-  if (param.sort_length == param.ref_length && records > param.max_rows)
+  if (param.rec_length == param.ref_length && records > param.max_rows)
     records=param.max_rows;			/* purecov: inspected */
 
   if (multi_byte_charset &&
@@ -127,9 +164,9 @@ ha_rows filesort(THD *thd, TABLE *table, SORT_FIELD *sortorder, uint s_length,
   while (memavl >= min_sort_memory)
   {
     ulong old_memavl;
-    ulong keys= memavl/(param.sort_length+sizeof(char*));
+    ulong keys= memavl/(param.rec_length+sizeof(char*));
     param.keys=(uint) min(records+1, keys);
-    if ((sort_keys= (uchar **) make_char_array(param.keys, param.sort_length,
+    if ((sort_keys= (uchar **) make_char_array(param.keys, param.rec_length,
 					       MYF(0))))
       break;
     old_memavl=memavl;
@@ -176,8 +213,8 @@ ha_rows filesort(THD *thd, TABLE *table, SORT_FIELD *sortorder, uint s_length,
       Use also the space previously used by string pointers in sort_buffer
       for temporary key storage.
     */
-    param.keys=((param.keys*(param.sort_length+sizeof(char*))) /
-		param.sort_length-1);
+    param.keys=((param.keys*(param.rec_length+sizeof(char*))) /
+		param.rec_length-1);
     maxbuffer--;				// Offset from 0
     if (merge_many_buff(&param,(uchar*) sort_keys,buffpek,&maxbuffer,
 			&tempfile))
@@ -356,8 +393,8 @@ static ha_rows find_all_keys(SORTPARAM *param, SQL_SELECT *select,
 	if (write_keys(param,sort_keys,idx,buffpek_pointers,tempfile))
 	  DBUG_RETURN(HA_POS_ERROR);
 	idx=0;
-	if (param->ref_length == param->sort_length &&
-	    my_b_tell(tempfile)/param->sort_length >= param->max_rows)
+	if (param->ref_length == param->rec_length &&
+	    my_b_tell(tempfile)/param->rec_length >= param->max_rows)
 	{
 	  /*
 	    We are writing the result index file and have found all
@@ -385,7 +422,7 @@ static ha_rows find_all_keys(SORTPARAM *param, SQL_SELECT *select,
       write_keys(param,sort_keys,idx,buffpek_pointers,tempfile))
     DBUG_RETURN(HA_POS_ERROR);			/* purecov: inspected */
   DBUG_RETURN(my_b_inited(tempfile) ?
-	      (ha_rows) (my_b_tell(tempfile)/param->sort_length) :
+	      (ha_rows) (my_b_tell(tempfile)/param->rec_length) :
 	      idx);
 } /* find_all_keys */
 
@@ -394,29 +431,30 @@ static ha_rows find_all_keys(SORTPARAM *param, SQL_SELECT *select,
 
 static int
 write_keys(SORTPARAM *param, register uchar **sort_keys, uint count,
-	   IO_CACHE *buffpek_pointers, IO_CACHE *tempfile)
+           IO_CACHE *buffpek_pointers, IO_CACHE *tempfile)
 {
-  uint sort_length;
+  uint sort_length, rec_length;
   uchar **end;
   BUFFPEK buffpek;
   DBUG_ENTER("write_keys");
 
-  sort_length=param->sort_length;
+  sort_length= param->sort_length;
+  rec_length= param->rec_length;
 #ifdef MC68000
   quicksort(sort_keys,count,sort_length);
 #else
-  my_string_ptr_sort((gptr) sort_keys,(uint) count,sort_length);
+  my_string_ptr_sort((gptr) sort_keys, (uint) count, sort_length);
 #endif
   if (!my_b_inited(tempfile) &&
-      open_cached_file(tempfile,mysql_tmpdir,TEMP_PREFIX,DISK_BUFFER_SIZE,
-			MYF(MY_WME)))
-    goto err;					/* purecov: inspected */
-  buffpek.file_pos=my_b_tell(tempfile);
+      open_cached_file(tempfile, mysql_tmpdir, TEMP_PREFIX, DISK_BUFFER_SIZE,
+                       MYF(MY_WME)))
+    goto err;                                        /* purecov: inspected */
+  buffpek.file_pos= my_b_tell(tempfile);
   if ((ha_rows) count > param->max_rows)
-    count=(uint) param->max_rows;		/* purecov: inspected */
+    count=(uint) param->max_rows;                /* purecov: inspected */
   buffpek.count=(ha_rows) count;
   for (end=sort_keys+count ; sort_keys != end ; sort_keys++)
-    if (my_b_write(tempfile,(byte*) *sort_keys,(uint) sort_length))
+    if (my_b_write(tempfile, (byte*) *sort_keys, (uint) rec_length))
       goto err;
   if (my_b_write(buffpek_pointers, (byte*) &buffpek, sizeof(buffpek)))
     goto err;
@@ -505,10 +543,10 @@ static void make_sortkey(register SORTPARAM *param,
           }
           else
           {
-	    my_strnxfrm(cs,(uchar*)to,length,(const uchar*)res->ptr(),length);
-	    bzero((char *)to+length,diff);
+             my_strnxfrm(cs,(uchar*)to,length,(const uchar*)res->ptr(),length);
+             bzero((char *)to+length,diff);
           }
-	  break;
+		break;
 	}
       case INT_RESULT:
 	{
@@ -577,29 +615,56 @@ static void make_sortkey(register SORTPARAM *param,
     else
       to+= sort_field->length;
   }
-  memcpy((byte*) to,ref_pos,(size_s) param->ref_length);/* Save filepos last */
+
+  if (param->addon_field)
+  {
+    /* 
+      Save field values appended to sorted fields.
+      First null bit indicators are appended then field values follow.
+      In this implementation we use fixed layout for field values -
+      the same for all records.
+    */
+    SORT_ADDON_FIELD *addonf= param->addon_field;
+    uchar *nulls= to;
+    DBUG_ASSERT(addonf);
+    bzero((char *) nulls, addonf->offset);
+    to+= addonf->offset;
+    for ( ; (field= addonf->field) ; addonf++)
+    {
+      if (addonf->null_bit && field->is_null())
+        nulls[addonf->null_offset]|= addonf->null_bit;
+      else
+        field->pack((char *) to, field->ptr);
+      to+= addonf->length;
+    }
+  }
+  else
+  {
+    /* Save filepos last */
+    memcpy((byte*) to, ref_pos, (size_s) param->ref_length);
+  }
   return;
 }
 
 
 static bool save_index(SORTPARAM *param, uchar **sort_keys, uint count)
 {
-  uint offset,ref_length;
+  uint offset,res_length;
   byte *to;
   DBUG_ENTER("save_index");
 
-  my_string_ptr_sort((gptr) sort_keys,(uint) count,param->sort_length);
-  ref_length=param->ref_length;
-  offset=param->sort_length-ref_length;
+  my_string_ptr_sort((gptr) sort_keys, (uint) count, param->sort_length);
+  res_length= param->res_length;
+  offset= param->rec_length-res_length;
   if ((ha_rows) count > param->max_rows)
     count=(uint) param->max_rows;
-  if (!(to=param->sort_form->record_pointers=
-	(byte*) my_malloc(ref_length*count,MYF(MY_WME))))
-    DBUG_RETURN(1);				/* purecov: inspected */
-  for (uchar **end=sort_keys+count ; sort_keys != end ; sort_keys++)
+  if (!(to= param->sort_form->sort.record_pointers=
+        (byte*) my_malloc(res_length*count, MYF(MY_WME))))
+    DBUG_RETURN(1);                 /* purecov: inspected */
+  for (uchar **end= sort_keys+count ; sort_keys != end ; sort_keys++)
   {
-    memcpy(to,*sort_keys+offset,ref_length);
-    to+=ref_length;
+    memcpy(to, *sort_keys+offset, res_length);
+    to+= res_length;
   }
   DBUG_RETURN(0);
 }
@@ -654,7 +719,7 @@ int merge_many_buff(SORTPARAM *param, uchar *sort_buffer,
 	/* This returns (uint) -1 if something goes wrong */
 
 uint read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek,
-		    uint sort_length)
+		    uint rec_length)
 {
   register uint count;
   uint length;
@@ -662,33 +727,35 @@ uint read_to_buffer(IO_CACHE *fromfile, BUFFPEK *buffpek,
   if ((count=(uint) min((ha_rows) buffpek->max_keys,buffpek->count)))
   {
     if (my_pread(fromfile->file,(byte*) buffpek->base,
-		 (length= sort_length*count),buffpek->file_pos,MYF_RW))
+		 (length= rec_length*count),buffpek->file_pos,MYF_RW))
       return((uint) -1);			/* purecov: inspected */
     buffpek->key=buffpek->base;
     buffpek->file_pos+= length;			/* New filepos */
     buffpek->count-=	count;
     buffpek->mem_count= count;
   }
-  return (count*sort_length);
+  return (count*rec_length);
 } /* read_to_buffer */
 
 
-	/* Merge buffers to one buffer */
+/* 
+   Merge buffers to one buffer 
+*/
 
 int merge_buffers(SORTPARAM *param, IO_CACHE *from_file,
-		  IO_CACHE *to_file, uchar *sort_buffer,
-		  BUFFPEK *lastbuff, BUFFPEK *Fb, BUFFPEK *Tb,
-		  int flag)
+                  IO_CACHE *to_file, uchar *sort_buffer,
+                  BUFFPEK *lastbuff, BUFFPEK *Fb, BUFFPEK *Tb,
+                  int flag)
 {
   int error;
-  uint sort_length,offset;
+  uint rec_length,sort_length,res_length,offset;
   ulong maxcount;
   ha_rows max_rows,org_max_rows;
   my_off_t to_start_filepos;
   uchar *strpos;
   BUFFPEK *buffpek,**refpek;
   QUEUE queue;
-  qsort2_cmp    cmp;
+  qsort2_cmp cmp;
   volatile bool *killed= &current_thd->killed;
   bool not_killable;
   DBUG_ENTER("merge_buffers");
@@ -697,29 +764,32 @@ int merge_buffers(SORTPARAM *param, IO_CACHE *from_file,
   if (param->not_killable)
   {
     killed= &not_killable;
-    not_killable=0;
+    not_killable= 0;
   }
 
   error=0;
-  offset=(sort_length=param->sort_length)-param->ref_length;
-  maxcount=(ulong) (param->keys/((uint) (Tb-Fb) +1));
-  to_start_filepos=my_b_tell(to_file);
-  strpos=(uchar*) sort_buffer;
-  org_max_rows=max_rows=param->max_rows;
-
-  if (init_queue(&queue,(uint) (Tb-Fb)+1,offsetof(BUFFPEK,key),0,
-		 (queue_compare)
-		 (cmp=get_ptr_compare(sort_length)),(void*) &sort_length))
-    DBUG_RETURN(1);				/* purecov: inspected */
+  rec_length= param->rec_length;
+  res_length= param->res_length;
+  sort_length= param->sort_length;
+  offset= rec_length-res_length;
+  maxcount= (ulong) (param->keys/((uint) (Tb-Fb) +1));
+  to_start_filepos= my_b_tell(to_file);
+  strpos= (uchar*) sort_buffer;
+  org_max_rows=max_rows= param->max_rows;
+
+  if (init_queue(&queue, (uint) (Tb-Fb)+1, offsetof(BUFFPEK,key), 0,
+                 (queue_compare) (cmp= get_ptr_compare(sort_length)),
+                 (void*) &sort_length))
+    DBUG_RETURN(1);                                /* purecov: inspected */
   for (buffpek= Fb ; buffpek <= Tb ; buffpek++)
   {
     buffpek->base= strpos;
-    buffpek->max_keys=maxcount;
-    strpos+= (uint) (error=(int) read_to_buffer(from_file,buffpek,
-						sort_length));
+    buffpek->max_keys= maxcount;
+    strpos+= (uint) (error= (int) read_to_buffer(from_file, buffpek,
+                                                                         rec_length));
     if (error == -1)
-      goto err;					/* purecov: inspected */
-    queue_insert(&queue,(byte*) buffpek);
+      goto err;                                        /* purecov: inspected */
+    queue_insert(&queue, (byte*) buffpek);
   }
 
   if (param->unique_buff)
@@ -732,98 +802,101 @@ int merge_buffers(SORTPARAM *param, IO_CACHE *from_file,
        This is safe as we know that there is always more than one element
        in each block to merge (This is guaranteed by the Unique:: algorithm
     */
-    buffpek=(BUFFPEK*) queue_top(&queue);
-    memcpy(param->unique_buff, buffpek->key, sort_length);
-    if (my_b_write(to_file,(byte*) buffpek->key, sort_length))
+    buffpek= (BUFFPEK*) queue_top(&queue);
+    memcpy(param->unique_buff, buffpek->key, rec_length);
+    if (my_b_write(to_file, (byte*) buffpek->key, rec_length))
     {
-      error=1; goto err;			/* purecov: inspected */
+      error=1; goto err;                        /* purecov: inspected */
     }
-    buffpek->key+=sort_length;
+    buffpek->key+= rec_length;
     buffpek->mem_count--;
     if (!--max_rows)
     {
-      error=0;					/* purecov: inspected */
-      goto end;					/* purecov: inspected */
+      error= 0;                                       /* purecov: inspected */
+      goto end;                                       /* purecov: inspected */
     }
-    queue_replaced(&queue);			// Top element has been used
+    queue_replaced(&queue);                        // Top element has been used
   }
   else
-    cmp=0;					// Not unique
+    cmp= 0;                                        // Not unique
 
   while (queue.elements > 1)
   {
     if (*killed)
     {
-      error=1; goto err;                        /* purecov: inspected */
+      error= 1; goto err;                        /* purecov: inspected */
     }
     for (;;)
     {
-      buffpek=(BUFFPEK*) queue_top(&queue);
-      if (cmp)					// Remove duplicates
+      buffpek= (BUFFPEK*) queue_top(&queue);
+      if (cmp)                                        // Remove duplicates
       {
-	if (!(*cmp)(&sort_length, &(param->unique_buff),
-		    (uchar**) &buffpek->key))
-	  goto skip_duplicate;
-	memcpy(param->unique_buff, (uchar*) buffpek->key,sort_length);
+        if (!(*cmp)(&sort_length, &(param->unique_buff),
+                    (uchar**) &buffpek->key))
+              goto skip_duplicate;
+            memcpy(param->unique_buff, (uchar*) buffpek->key, rec_length);
       }
       if (flag == 0)
       {
-	if (my_b_write(to_file,(byte*) buffpek->key, sort_length))
-	{
-	  error=1; goto err;			/* purecov: inspected */
-	}
+        if (my_b_write(to_file,(byte*) buffpek->key, rec_length))
+        {
+          error=1; goto err;                        /* purecov: inspected */
+        }
       }
       else
       {
-	WRITE_REF(to_file,(byte*) buffpek->key+offset);
+        if (my_b_write(to_file, (byte*) buffpek->key+offset, res_length))
+        {
+          error=1; goto err;                        /* purecov: inspected */
+        }
       }
       if (!--max_rows)
       {
-	error=0;				/* purecov: inspected */
-	goto end;				/* purecov: inspected */
+        error= 0;                               /* purecov: inspected */
+        goto end;                               /* purecov: inspected */
       }
 
     skip_duplicate:
-      buffpek->key+=sort_length;
+      buffpek->key+= rec_length;
       if (! --buffpek->mem_count)
       {
-	if (!(error=(int) read_to_buffer(from_file,buffpek,
-					 sort_length)))
-	{
-	  uchar *base=buffpek->base;
-	  ulong max_keys=buffpek->max_keys;
-
-	  VOID(queue_remove(&queue,0));
-
-	  /* Put room used by buffer to use in other buffer */
-	  for (refpek= (BUFFPEK**) &queue_top(&queue);
-	       refpek <= (BUFFPEK**) &queue_end(&queue);
-	       refpek++)
-	  {
-	    buffpek= *refpek;
-	    if (buffpek->base+buffpek->max_keys*sort_length == base)
-	    {
-	      buffpek->max_keys+=max_keys;
-	      break;
-	    }
-	    else if (base+max_keys*sort_length == buffpek->base)
-	    {
-	      buffpek->base=base;
-	      buffpek->max_keys+=max_keys;
-	      break;
-	    }
-	  }
-	  break;			/* One buffer have been removed */
-	}
-	else if (error == -1)
-	  goto err;			/* purecov: inspected */
+        if (!(error= (int) read_to_buffer(from_file,buffpek,
+                                          rec_length)))
+        {
+          uchar *base= buffpek->base;
+          ulong max_keys= buffpek->max_keys;
+
+          VOID(queue_remove(&queue,0));
+
+          /* Put room used by buffer to use in other buffer */
+          for (refpek= (BUFFPEK**) &queue_top(&queue);
+               refpek <= (BUFFPEK**) &queue_end(&queue);
+               refpek++)
+          {
+            buffpek= *refpek;
+            if (buffpek->base+buffpek->max_keys*rec_length == base)
+            {
+              buffpek->max_keys+= max_keys;
+              break;
+            }
+            else if (base+max_keys*rec_length == buffpek->base)
+            {
+              buffpek->base= base;
+              buffpek->max_keys+= max_keys;
+              break;
+            }
+          }
+          break;                        /* One buffer have been removed */
+        }
+        else if (error == -1)
+          goto err;                        /* purecov: inspected */
       }
-      queue_replaced(&queue);		/* Top element has been replaced */
+      queue_replaced(&queue);              /* Top element has been replaced */
     }
   }
-  buffpek=(BUFFPEK*) queue_top(&queue);
+  buffpek= (BUFFPEK*) queue_top(&queue);
   buffpek->base= sort_buffer;
-  buffpek->max_keys=param->keys;
+  buffpek->max_keys= param->keys;
 
   /*
     As we know all entries in the buffer are unique, we only have to
@@ -833,7 +906,7 @@ int merge_buffers(SORTPARAM *param, IO_CACHE *from_file,
   {
     if (!(*cmp)(&sort_length, &(param->unique_buff), (uchar**) &buffpek->key))
     {
-      buffpek->key+=sort_length;	// Remove duplicate
+      buffpek->key+= rec_length;         // Remove duplicate
       --buffpek->mem_count;
     }
   }
@@ -841,37 +914,40 @@ int merge_buffers(SORTPARAM *param, IO_CACHE *from_file,
   do
   {
     if ((ha_rows) buffpek->mem_count > max_rows)
-    {					/* Don't write too many records */
-      buffpek->mem_count=(uint) max_rows;
-      buffpek->count=0;			/* Don't read more */
+    {                                        /* Don't write too many records */
+      buffpek->mem_count= (uint) max_rows;
+      buffpek->count= 0;                        /* Don't read more */
     }
-    max_rows-=buffpek->mem_count;
+    max_rows-= buffpek->mem_count;
     if (flag == 0)
     {
       if (my_b_write(to_file,(byte*) buffpek->key,
-		     (sort_length*buffpek->mem_count)))
+                     (rec_length*buffpek->mem_count)))
       {
-	error=1; goto err;			/* purecov: inspected */
+        error= 1; goto err;                        /* purecov: inspected */
       }
     }
     else
     {
       register uchar *end;
       strpos= buffpek->key+offset;
-      for (end=strpos+buffpek->mem_count*sort_length;
-	   strpos != end ;
-	   strpos+=sort_length)
-      {
-	WRITE_REF(to_file,strpos);
+      for (end= strpos+buffpek->mem_count*rec_length ;
+           strpos != end ;
+           strpos+= rec_length)
+      {     
+        if (my_b_write(to_file, (byte *) strpos, res_length))
+        {
+          error=1; goto err;                        
+        }
       }
     }
   }
-  while ((error=(int) read_to_buffer(from_file,buffpek,sort_length))
-	 != -1 && error != 0);
+  while ((error=(int) read_to_buffer(from_file,buffpek, rec_length))
+         != -1 && error != 0);
 
 end:
-  lastbuff->count=min(org_max_rows-max_rows,param->max_rows);
-  lastbuff->file_pos=to_start_filepos;
+  lastbuff->count= min(org_max_rows-max_rows, param->max_rows);
+  lastbuff->file_pos= to_start_filepos;
 err:
   delete_queue(&queue);
   DBUG_RETURN(error);
@@ -925,7 +1001,6 @@ sortlength(SORT_FIELD *sortorder, uint s_length, bool *multi_byte_charset)
     sortorder->need_strxnfrm= 0;
     if (sortorder->field)
     {
-      
       if (sortorder->field->type() == FIELD_TYPE_BLOB)
 	sortorder->length= thd->variables.max_sort_length;
       else
@@ -947,7 +1022,7 @@ sortlength(SORT_FIELD *sortorder, uint s_length, bool *multi_byte_charset)
       case STRING_RESULT:
 	sortorder->length=sortorder->item->max_length;
 	if (use_strnxfrm((cs=sortorder->item->charset())))
-	{
+	{ 
 	  sortorder->length= sortorder->length*cs->strxfrm_multiply;
 	  sortorder->need_strxnfrm= 1;
 	  *multi_byte_charset= 1;
@@ -982,6 +1057,148 @@ sortlength(SORT_FIELD *sortorder, uint s_length, bool *multi_byte_charset)
 
 
 /*
+  Get descriptors of fields appended to sorted fields and
+  calculate its total length
+
+  SYNOPSIS
+    get_addon_fields()
+    thd                 Current thread
+    ptabfields          Array of references to the table fields
+    sortlength          Total length of sorted fields
+    plength    out:     Total length of appended fields
+
+  DESCRIPTION
+    The function first finds out what fields are used in the result set.
+    Then it calculates the length of the buffer to store the values of
+    these fields together with the value of sort values. 
+    If the calculated length is not greater than max_length_for_sort_data
+    the function allocates memory for an array of descriptors containing
+    layouts for the values of the non-sorted fields in the buffer and
+    fills them.
+
+  NOTES
+    The null bits for the appended values are supposed to be put together
+    and stored the buffer just ahead of the value of the first field.
+
+  RETURN
+    Pointer to the layout descriptors for the appended fields, if any
+    NULL - if we do not store field values with sort data.
+*/
+
+static SORT_ADDON_FIELD *
+get_addon_fields(THD *thd, Field **ptabfield, uint sortlength, uint *plength)
+{
+  Field **pfield;
+  Field *field;
+  SORT_ADDON_FIELD *addonf;
+  uint length= 0;
+  uint fields= 0;
+  uint null_fields= 0;
+
+  /* 
+     If there is a reference to a field in the query add it
+     to the the set of appended fields.
+     Note for future refinement:
+     This this a too strong condition.
+     Actually we need only the fields referred in the
+     result set. And for some of them it makes sense to use 
+     the values directly from sorted fields.
+  */
+  *plength= 0;
+  /*
+     The following statement is added to avoid sorting in alter_table.
+     The fact is the filter 'field->query_id != thd->query_id'
+     doesn't work for alter table
+  */
+  if (thd->lex.sql_command != SQLCOM_SELECT)
+    return 0;
+  for (pfield= ptabfield; (field= *pfield) ; pfield++)
+  {
+    if (field->query_id != thd->query_id)
+      continue;
+    if (field->flags & BLOB_FLAG)
+      return 0;
+    length+= field->max_packed_col_length(field->pack_length());
+    if (field->maybe_null())
+      null_fields++;
+    fields++;
+  } 
+  if (!fields)
+    return 0;
+  length+= (null_fields+7)/8;
+
+  if (length+sortlength > thd->variables.max_length_for_sort_data ||
+      !(addonf= (SORT_ADDON_FIELD *) my_malloc(sizeof(SORT_ADDON_FIELD)*
+                                               (fields+1), MYF(MY_WME))))
+    return 0;
+
+  *plength= length;
+  length= (null_fields+7)/8;
+  null_fields= 0;
+  for (pfield= ptabfield; (field= *pfield) ; pfield++)
+  {
+    if (field->query_id != thd->query_id)
+      continue;
+    addonf->field= field;
+    addonf->offset= length;
+    if (field->maybe_null())
+    {
+      addonf->null_offset= null_fields/8;
+      addonf->null_bit= 1<<(null_fields & 7);
+      null_fields++;
+    }
+    else
+    {
+      addonf->null_offset= 0;
+      addonf->null_bit= 0;
+    }
+    addonf->length= field->max_packed_col_length(field->pack_length());
+    length+= addonf->length;
+    addonf++;
+  }
+  addonf->field= 0;     // Put end marker
+  
+  DBUG_PRINT("info",("addon_length: %d",length));
+  return (addonf-fields);
+}
+
+
+/*
+  Copy (unpack) values appended to sorted fields from a buffer back to 
+  their regular positions specified by the Field::ptr pointers.
+
+  SYNOPSIS
+    unpack_addon_fields()
+    addon_field     Array of descriptors for appended fields
+    buff            Buffer which to unpack the value from
+
+  NOTES
+    The function is supposed to be used only as a callback function
+    when getting field values for the sorted result set.
+
+  RETURN
+    void.
+*/
+
+static void 
+unpack_addon_fields(struct st_sort_addon_field *addon_field, byte *buff)
+{
+  Field *field;
+  SORT_ADDON_FIELD *addonf= addon_field;
+
+  for ( ; (field= addonf->field) ; addonf++)
+  {
+    if (addonf->null_bit && (addonf->null_bit & buff[addonf->null_offset]))
+    {
+      field->set_null();
+      continue;
+    }
+    field->set_notnull();
+    field->unpack(field->ptr, (char *) buff+addonf->offset);
+  }
+}
+
+/*
 ** functions to change a double or float to a sortable string
 ** The following should work for IEEE
 */
diff --git a/sql/ha_berkeley.h b/sql/ha_berkeley.h
index dd42e7ab9e2..f1669e9b6c7 100644
--- a/sql/ha_berkeley.h
+++ b/sql/ha_berkeley.h
@@ -166,6 +166,7 @@ class ha_berkeley: public handler
   }
   longlong get_auto_increment();
   void print_error(int error, myf errflag);
+  uint8 table_cache_type() { return HA_CACHE_TBL_TRANSACT; }
 };
 
 extern bool berkeley_skip, berkeley_shared_data;
diff --git a/sql/ha_innodb.cc b/sql/ha_innodb.cc
index ce26ee705dd..4f955d8f79e 100644
--- a/sql/ha_innodb.cc
+++ b/sql/ha_innodb.cc
@@ -475,7 +475,7 @@ If thd is not in the autocommit state, this function also starts a new
 transaction for thd if there is no active trx yet, and assigns a consistent
 read view to it if there is no read view yet. */
 
-my_bool
+bool
 innobase_query_caching_of_table_permitted(
 /*======================================*/
 				/* out: TRUE if permitted, FALSE if not;
@@ -916,11 +916,11 @@ innobase_commit_low(
                 /* Update the replication position info inside InnoDB */
 
                 trx->mysql_master_log_file_name
-                                        = active_mi->rli.master_log_name;
+                                        = active_mi->rli.group_master_log_name;
                 trx->mysql_master_log_pos = ((ib_longlong)
-					 (active_mi->rli.master_log_pos +
-					  active_mi->rli.event_len +
-					  active_mi->rli.pending));
+					 (active_mi->rli.group_master_log_pos +
+					  active_mi->rli.event_len
+                                          ));
         }
 #endif /* HAVE_REPLICATION */
         trx_commit_for_mysql(trx);
diff --git a/sql/ha_innodb.h b/sql/ha_innodb.h
index 56546e3e8d0..be1174added 100644
--- a/sql/ha_innodb.h
+++ b/sql/ha_innodb.h
@@ -168,6 +168,7 @@ class ha_innobase: public handler
 			     		enum thr_lock_type lock_type);
 	void init_table_handle_for_HANDLER(); 
 	longlong get_auto_increment();
+	uint8 table_cache_type() { return HA_CACHE_TBL_ASKTRANSACT; }
 };
 
 extern bool innodb_skip;
@@ -207,6 +208,6 @@ int innobase_close_connection(THD *thd);
 int innobase_drop_database(char *path);
 int innodb_show_status(THD* thd);
 
-my_bool innobase_query_caching_of_table_permitted(THD* thd, char* full_name,
-						uint full_name_len);
+bool innobase_query_caching_of_table_permitted(THD* thd, char* full_name,
+					       uint full_name_len);
 void innobase_release_temporary_latches(void* innobase_tid);
diff --git a/sql/handler.cc b/sql/handler.cc
index e288d590e88..762e07eec64 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -877,6 +877,16 @@ int handler::delete_all_rows()
   return (my_errno=HA_ERR_WRONG_COMMAND);
 }
 
+bool handler::caching_allowed(THD* thd, char* table_key, 
+			      uint key_length, uint8 cache_type)
+{
+  if (cache_type == HA_CACHE_TBL_ASKTRANSACT)
+    return innobase_query_caching_of_table_permitted(thd, table_key,
+						     key_length);
+  else
+    return 1;
+}
+
 /****************************************************************************
 ** Some general functions that isn't in the handler class
 ****************************************************************************/
diff --git a/sql/handler.h b/sql/handler.h
index 06c24214d0f..8c23a3625e0 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -117,6 +117,11 @@
 #define HA_OPTION_NO_DELAY_KEY_WRITE (1L << 18)
 #define HA_MAX_REC_LENGTH	65535
 
+/* Table caching type */
+#define HA_CACHE_TBL_NONTRANSACT 0
+#define HA_CACHE_TBL_ASKTRANSACT 1
+#define HA_CACHE_TBL_TRANSACT    2
+
 enum db_type { DB_TYPE_UNKNOWN=0,DB_TYPE_DIAB_ISAM=1,
 	       DB_TYPE_HASH,DB_TYPE_MISAM,DB_TYPE_PISAM,
 	       DB_TYPE_RMS_ISAM, DB_TYPE_HEAP, DB_TYPE_ISAM,
@@ -343,6 +348,16 @@ public:
   virtual THR_LOCK_DATA **store_lock(THD *thd,
 				     THR_LOCK_DATA **to,
 				     enum thr_lock_type lock_type)=0;
+
+  /* Type of table for caching query */
+  virtual uint8 table_cache_type() { return HA_CACHE_TBL_NONTRANSACT; }
+  /* 
+    Is query with this cable cachable (have sense only for ASKTRANSACT 
+    tables)
+  */
+  static bool caching_allowed(THD* thd, char* table_key, 
+			      uint key_length, uint8 cahe_type);
+
 };
 
 	/* Some extern variables used with handlers */
diff --git a/sql/item.cc b/sql/item.cc
index 60d52e776fd..7dd8392b695 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -172,7 +172,7 @@ bool Item::get_time(TIME *ltime)
 
 CHARSET_INFO * Item::default_charset() const
 {
-  return current_thd->db_charset;
+  return current_thd->variables.collation_connection;
 }
 
 bool Item::set_charset(CHARSET_INFO *cs1, enum coercion co1,
diff --git a/sql/item.h b/sql/item.h
index 6a7ebd506ac..4862ad21fbe 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -603,6 +603,7 @@ public:
     item(it) 
   {}
   bool fix_fields(THD *, struct st_table_list *, Item ** ref);
+  Item **storage() {return &item;}
 };
 
 /*
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 09f0aeefb09..0f21cf5a774 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -1781,6 +1781,43 @@ longlong Item_func_isnull::val_int()
   return args[0]->is_null() ? 1: 0;
 }
 
+longlong Item_is_not_null_test::val_int()
+{
+  DBUG_ENTER("Item_is_not_null_test::val_int");
+  if (!used_tables_cache)
+  {
+    owner->was_null|= (!cached_value);
+    DBUG_PRINT("info", ("cached :%d", cached_value));
+    DBUG_RETURN(cached_value);
+  }
+  if (args[0]->is_null())
+  {
+    DBUG_PRINT("info", ("null"))
+    owner->was_null|= 1;
+    DBUG_RETURN(0);
+  }
+  else
+    DBUG_RETURN(1);
+}
+
+/* Optimize case of not_null_column IS NULL */
+void Item_is_not_null_test::update_used_tables()
+{
+  if (!args[0]->maybe_null)
+  {
+    used_tables_cache= 0;			/* is always true */
+    cached_value= (longlong) 1;
+  }
+  else
+  {
+    args[0]->update_used_tables();
+    if (!(used_tables_cache=args[0]->used_tables()))
+    {
+      /* Remember if the value is always NULL or never NULL */
+      cached_value= (longlong) !args[0]->is_null();
+    }
+  }
+}
 
 longlong Item_func_isnotnull::val_int()
 {
@@ -1870,7 +1907,7 @@ bool Item_func_like::fix_fields(THD *thd, TABLE_LIST *tlist, Item ** ref)
     {
       const char* tmp = first + 1;
       for (; *tmp != wild_many && *tmp != wild_one && *tmp != escape; tmp++) ;
-      canDoTurboBM = tmp == last;
+      canDoTurboBM = (tmp == last) && !use_mb(args[0]->charset());
     }
 
     if (canDoTurboBM)
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index e7670755396..5e246e3e285 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -648,6 +648,7 @@ class Item_func_in :public Item_int_func
 
 class Item_func_isnull :public Item_bool_func
 {
+protected:
   longlong cached_value;
 public:
   Item_func_isnull(Item *a) :Item_bool_func(a) {}
@@ -656,11 +657,11 @@ public:
   void fix_length_and_dec()
   {
     decimals=0; max_length=1; maybe_null=0;
-    Item_func_isnull::update_used_tables();
+    update_used_tables();
   }
   const char *func_name() const { return "isnull"; }
   /* Optimize case of not_null_column IS NULL */
-  void update_used_tables()
+  virtual void update_used_tables()
   {
     if (!args[0]->maybe_null)
     {
@@ -680,6 +681,22 @@ public:
   optimize_type select_optimize() const { return OPTIMIZE_NULL; }
 };
 
+/* Functions used by HAVING for rewriting IN subquery */
+
+class Item_in_subselect;
+class Item_is_not_null_test :public Item_func_isnull
+{
+  Item_in_subselect* owner;
+public:
+  Item_is_not_null_test(Item_in_subselect* ow, Item *a)
+    :Item_func_isnull(a), owner(ow)
+  {}
+  longlong val_int();
+  const char *func_name() const { return "is_not_null_test"; }
+  void update_used_tables();
+};
+
+
 class Item_func_isnotnull :public Item_bool_func
 {
 public:
diff --git a/sql/item_create.cc b/sql/item_create.cc
index 2fb753912eb..23e4ce1d2b4 100644
--- a/sql/item_create.cc
+++ b/sql/item_create.cc
@@ -646,7 +646,13 @@ Item *create_func_point(Item *a, Item *b)
   return new Item_func_point(a, b);
 }
 
-#ifdef HAVE_COMPRESS
+#if !defined(HAVE_COMPRESS)
+
+Item *create_func_compress           (Item*a __attribute__((unused))){return 0;}
+Item *create_func_uncompress         (Item*a __attribute__((unused))){return 0;}
+Item *create_func_uncompressed_length(Item*a __attribute__((unused))){return 0;}
+
+#else
 
 Item *create_func_compress(Item* a)
 {
diff --git a/sql/item_create.h b/sql/item_create.h
index 90595859bcc..c79fe07b8d4 100644
--- a/sql/item_create.h
+++ b/sql/item_create.h
@@ -142,9 +142,7 @@ Item *create_func_numgeometries(Item *a);
 
 Item *create_func_point(Item *a, Item *b);
 
-#ifdef HAVE_COMPRESS
 Item *create_func_compress(Item *a);
 Item *create_func_uncompress(Item *a);
 Item *create_func_uncompressed_length(Item *a);
-#endif
 
diff --git a/sql/item_func.cc b/sql/item_func.cc
index 7a01ea898bb..d427e3c5a3a 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -993,7 +993,8 @@ longlong Item_func_uncompressed_length::val_int()
     return 0; /* purecov: inspected */
   }
   null_value=0;
-  return uint4korr(res->c_ptr());
+  if (res->is_empty()) return 0;
+  return uint4korr(res->c_ptr()) & 0x3FFFFFFF;
 }
 
 #endif /* HAVE_COMPRESS */
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 82463c91a2f..bba3799d398 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -2925,6 +2925,8 @@ ret:
 String *Item_func_compress::val_str(String *str)
 {
   String *res= args[0]->val_str(str);
+  if (res->is_empty()) return res;
+
   int err= Z_OK;
   int code;
 
@@ -2940,14 +2942,13 @@ String *Item_func_compress::val_str(String *str)
    compress(compress(compress(...)))
    I.e. zlib give number 'at least'..
   */
-  uLongf new_size= (uLongf)((res->length()*120)/100)+12;
-
-  buffer.realloc((uint32)new_size+sizeof(int32)+sizeof(char));
+  ulong new_size= (ulong)((res->length()*120)/100)+12;
 
-  Byte *body= ((Byte*)buffer.c_ptr())+sizeof(int32);
-  err= compress(body, &new_size,(const Bytef*)res->c_ptr(), res->length());
+  buffer.realloc((uint32)new_size + 4 + 1);
+  Byte *body= ((Byte*)buffer.c_ptr()) + 4;
   
-  if (err != Z_OK)
+  if ((err= compress(body, &new_size,
+		     (const Bytef*)res->c_ptr(), res->length())) != Z_OK)
   {
     code= err==Z_MEM_ERROR ? ER_ZLIB_Z_MEM_ERROR : ER_ZLIB_Z_BUF_ERROR;
     push_warning(current_thd,MYSQL_ERROR::WARN_LEVEL_ERROR,code,ER(code));
@@ -2955,18 +2956,17 @@ String *Item_func_compress::val_str(String *str)
     return 0;
   }
   
-  int4store(buffer.c_ptr(),res->length());
-  buffer.length((uint32)new_size+sizeof(int32));
-  
-  /* This is for the stupid char fields which trimm ' ': */
+  int4store(buffer.c_ptr(),res->length() & 0x3FFFFFFF);
+
+  /* This is for the stupid char fields which trim ' ': */
   char *last_char= ((char*)body)+new_size-1;
   if (*last_char == ' ')
   {
     *++last_char= '.';
     new_size++;
   }
-  
-  buffer.length((uint32)new_size+sizeof(int32));
+
+  buffer.length((uint32)new_size + 4);
   
   return &buffer;
 }
@@ -2974,7 +2974,9 @@ String *Item_func_compress::val_str(String *str)
 String *Item_func_uncompress::val_str(String *str)
 {
   String *res= args[0]->val_str(str);
-  uLongf new_size= uint4korr(res->c_ptr());
+  if (res->is_empty()) return res;
+
+  ulong new_size= uint4korr(res->c_ptr()) & 0x3FFFFFFF;
   int err= Z_OK;
   uint code;
   
@@ -2983,16 +2985,14 @@ String *Item_func_uncompress::val_str(String *str)
     push_warning_printf(current_thd,MYSQL_ERROR::WARN_LEVEL_ERROR,
 			ER_TOO_BIG_FOR_UNCOMPRESS,
 			ER(ER_TOO_BIG_FOR_UNCOMPRESS),MAX_BLOB_WIDTH);
-    null_value= 1;
+    null_value= 0;
     return 0;
   }
   
   buffer.realloc((uint32)new_size);
   
-  err= uncompress((Byte*)buffer.c_ptr(), &new_size, 
-		  ((const Bytef*)res->c_ptr())+sizeof(int32),res->length());
-  
-  if (err == Z_OK)
+  if ((err= uncompress((Byte*)buffer.c_ptr(), &new_size, 
+		       ((const Bytef*)res->c_ptr())+4,res->length())) == Z_OK)
   {
     buffer.length((uint32)new_size);
     return &buffer;
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 76451680b59..6c0b799b4de 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -508,9 +508,31 @@ void Item_in_subselect::single_value_transformer(THD *thd,
       sl->item_list.push_back(new Item_int("Not_used", (longlong) 1, 21));
       if (sl->table_list.elements)
       {
-	item= (*func)(expr, new Item_asterisk_remover(this, item,
-						      (char *)"<no matter>",
-						      (char*)"<result>"));
+	Item *having= item, *isnull= item;
+	if (item->type() == Item::FIELD_ITEM &&
+	    ((Item_field*) item)->field_name[0] == '*')
+	{
+	  Item_asterisk_remover *remover;
+	  item= remover= new Item_asterisk_remover(this, item,
+						   (char*)"<no matter>",
+						   (char*)"<result>");
+	  having= 
+	    new Item_is_not_null_test(this,
+				      new Item_ref(remover->storage(),
+						   (char*)"<no matter>",
+						   (char*)"<null test>"));
+	  isnull=
+	    new Item_is_not_null_test(this,
+				      new Item_ref(remover->storage(),
+						   (char*)"<no matter>",
+						   (char*)"<null test>"));
+	}
+	having= new Item_is_not_null_test(this, having);
+	sl->having= (sl->having ?
+		     new Item_cond_and(having, sl->having) :
+		     having);
+	item= new Item_cond_or((*func)(expr, item),
+			       new Item_func_isnull(isnull));
 	sl->where= and_items(sl->where, item);
       }
       else
diff --git a/sql/item_subselect.h b/sql/item_subselect.h
index 351c4af7f33..fc4dad5a6b3 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -183,6 +183,7 @@ public:
 
   friend class Item_asterisk_remover;
   friend class Item_ref_null_helper;
+  friend class Item_is_not_null_test;
 };
 
 /* ALL/ANY/SOME subselect */
diff --git a/sql/lex.h b/sql/lex.h
index b5a81a30991..e51b3efff87 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -448,9 +448,7 @@ static SYMBOL sql_functions[] = {
   { "CHARACTER_LENGTH", SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_char_length)},
   { "COALESCE",		SYM(COALESCE),0,0},
   { "COERCIBILITY",	SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_coercibility)},
-#ifdef HAVE_COMPRESS
   { "COMPRESS",		SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_compress)},
-#endif
   { "CONCAT",		SYM(CONCAT),0,0},
   { "CONCAT_WS",	SYM(CONCAT_WS),0,0},
   { "CONNECTION_ID",	SYM(FUNC_ARG0),0,CREATE_FUNC(create_func_connection_id)},
@@ -627,10 +625,8 @@ static SYMBOL sql_functions[] = {
   { "TOUCHES",		SYM(FUNC_ARG2),0,CREATE_FUNC(create_func_touches)},
   { "TRIM",		SYM(TRIM),0,0},
   { "UCASE",		SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_ucase)},
-#ifdef HAVE_COMPRESS
   { "UNCOMPRESS",	SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_uncompress)},
   { "UNCOMPRESSED_LENGTH", SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_uncompressed_length)},
-#endif
   { "UNIQUE_USERS",	SYM(UNIQUE_USERS),0,0},
   { "UNIX_TIMESTAMP",	SYM(UNIX_TIMESTAMP),0,0},
   { "UPPER",		SYM(FUNC_ARG1),0,CREATE_FUNC(create_func_ucase)},
diff --git a/sql/log.cc b/sql/log.cc
index 51b1c572601..50471041ee1 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -592,24 +592,32 @@ err:
 
 
 /*
-  Delete the current log file, remove it from index file and start on next 
+  Delete relay log files prior to rli->group_relay_log_name
+  (i.e. all logs which are not involved in a non-finished group
+  (transaction)), remove them from the index file and start on next relay log.
 
   SYNOPSIS
     purge_first_log()
-    rli		Relay log information
-
+    rli		 Relay log information
+    included     If false, all relay logs that are strictly before
+                 rli->group_relay_log_name are deleted ; if true, the latter is
+                 deleted too (i.e. all relay logs
+                 read by the SQL slave thread are deleted).
+    
   NOTE
     - This is only called from the slave-execute thread when it has read
-      all commands from a log and want to switch to a new log.
-    - When this happens, we should never be in an active transaction as
-      a transaction is always written as a single block to the binary log.
+      all commands from a relay log and want to switch to a new relay log.
+    - When this happens, we can be in an active transaction as
+      a transaction can span over two relay logs
+      (although it is always written as a single block to the master's binary 
+      log, hence cannot span over two master's binary logs).
 
   IMPLEMENTATION
     - Protects index file with LOCK_index
-    - Delete first log file,
-    - Copy all file names after this one to the front of the index file
+    - Delete relevant relay log files
+    - Copy all file names after these ones to the front of the index file
     - If the OS has truncate, truncate the file, else fill it with \n'
-    - Read the first file name from the index file and store in rli->linfo
+    - Read the next file name from the index file and store in rli->linfo
 
   RETURN VALUES
     0			ok
@@ -620,66 +628,68 @@ err:
 
 #ifdef HAVE_REPLICATION
 
-int MYSQL_LOG::purge_first_log(struct st_relay_log_info* rli)
+int MYSQL_LOG::purge_first_log(struct st_relay_log_info* rli, bool included) 
 {
   int error;
   DBUG_ENTER("purge_first_log");
 
-  /*
-    Test pre-conditions.
-
-    Assume that we have previously read the first log and
-    stored it in rli->relay_log_name
-  */
   DBUG_ASSERT(is_open());
   DBUG_ASSERT(rli->slave_running == 1);
-  DBUG_ASSERT(!strcmp(rli->linfo.log_file_name,rli->relay_log_name));
-  DBUG_ASSERT(rli->linfo.index_file_offset ==
-	      strlen(rli->relay_log_name) + 1);
+  DBUG_ASSERT(!strcmp(rli->linfo.log_file_name,rli->event_relay_log_name));
 
-  /* We have already processed the relay log, so it's safe to delete it */
-  my_delete(rli->relay_log_name, MYF(0));
   pthread_mutex_lock(&LOCK_index);
-  if (copy_up_file_and_fill(&index_file, rli->linfo.index_file_offset))
-  {
-    error= LOG_INFO_IO;
-    goto err;
-  }
+  pthread_mutex_lock(&rli->log_space_lock);
+  rli->relay_log.purge_logs(rli->group_relay_log_name, included,
+                            0, 0, &rli->log_space_total);
+  // Tell the I/O thread to take the relay_log_space_limit into account
+  rli->ignore_log_space_limit= 0;
+  pthread_mutex_unlock(&rli->log_space_lock);
 
   /*
-    Update the space counter used by all relay logs
     Ok to broadcast after the critical region as there is no risk of
     the mutex being destroyed by this thread later - this helps save
     context switches
   */
-  pthread_mutex_lock(&rli->log_space_lock);
-  rli->log_space_total -= rli->relay_log_pos;
-  //tell the I/O thread to take the relay_log_space_limit into account
-  rli->ignore_log_space_limit= 0;
-  pthread_mutex_unlock(&rli->log_space_lock);
   pthread_cond_broadcast(&rli->log_space_cond);
   
   /*
     Read the next log file name from the index file and pass it back to
     the caller
+    If included is true, we want the first relay log;
+    otherwise we want the one after event_relay_log_name.
   */
-  if ((error=find_log_pos(&rli->linfo, NullS, 0 /*no mutex*/)))
+  if ((included && (error=find_log_pos(&rli->linfo, NullS, 0))) ||
+      (!included &&
+       ((error=find_log_pos(&rli->linfo, rli->event_relay_log_name, 0)) ||
+        (error=find_next_log(&rli->linfo, 0)))))
   {
     char buff[22];
-    sql_print_error("next log error: %d  offset: %s  log: %s",
-		    error,
-		    llstr(rli->linfo.index_file_offset,buff),
-		    rli->linfo.log_file_name);
+    sql_print_error("next log error: %d  offset: %s  log: %s included: %d",
+                    error,
+                    llstr(rli->linfo.index_file_offset,buff),
+                    rli->group_relay_log_name,
+                    included);
     goto err;
   }
+
   /*
-    Reset position to current log.  This involves setting both of the
-    position variables:
+    Reset rli's coordinates to the current log.
   */
-  rli->relay_log_pos = BIN_LOG_HEADER_SIZE;
-  rli->pending = 0;
-  strmake(rli->relay_log_name,rli->linfo.log_file_name,
-	  sizeof(rli->relay_log_name)-1);
+  rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE;
+  strmake(rli->event_relay_log_name,rli->linfo.log_file_name,
+	  sizeof(rli->event_relay_log_name)-1);
+
+  /*
+    If we removed the rli->group_relay_log_name file,
+    we must update the rli->group* coordinates, otherwise do not touch it as the
+    group's execution is not finished (e.g. COMMIT not executed)
+  */
+  if (included)
+  {
+    rli->group_relay_log_pos = BIN_LOG_HEADER_SIZE;
+    strmake(rli->group_relay_log_name,rli->linfo.log_file_name,
+            sizeof(rli->group_relay_log_name)-1);
+  }
 
   /* Store where we are in the new file for the execution thread */
   flush_relay_log_info(rli);
@@ -693,13 +703,14 @@ err:
   Update log index_file
 */
 
-int MYSQL_LOG::update_log_index(LOG_INFO* log_info)
+int MYSQL_LOG::update_log_index(LOG_INFO* log_info, bool need_update_threads)
 {
   if (copy_up_file_and_fill(&index_file, log_info->index_file_start_offset))
     return LOG_INFO_IO;
 
   // now update offsets in index file for running threads
-  adjust_linfo_offsets(log_info->index_file_start_offset);
+  if (need_update_threads)
+    adjust_linfo_offsets(log_info->index_file_start_offset);
   return 0;
 }
 
@@ -708,9 +719,13 @@ int MYSQL_LOG::update_log_index(LOG_INFO* log_info)
 
   SYNOPSIS
     purge_logs()
-    thd		Thread pointer
-    to_log	Delete all log file name before this file. This file is not
-		deleted
+    to_log	        Delete all log file name before this file. 
+    included            If true, to_log is deleted too.
+    need_mutex
+    need_update_threads If we want to update the log coordinates of
+                        all threads. False for relay logs, true otherwise.
+    freed_log_space     If not null, decrement this variable of
+                        the amount of log space freed
 
   NOTES
     If any of the logs before the deleted one is in use,
@@ -722,31 +737,59 @@ int MYSQL_LOG::update_log_index(LOG_INFO* log_info)
     LOG_INFO_EOF		to_log not found
 */
 
-int MYSQL_LOG::purge_logs(THD* thd, const char* to_log)
+int MYSQL_LOG::purge_logs(const char *to_log, 
+                          bool included,
+                          bool need_mutex, 
+                          bool need_update_threads, 
+                          ulonglong *decrease_log_space)
 {
   int error;
+  bool exit_loop= 0;
   LOG_INFO log_info;
   DBUG_ENTER("purge_logs");
+  DBUG_PRINT("info",("to_log= %s",to_log));
 
   if (no_rotate)
     DBUG_RETURN(LOG_INFO_PURGE_NO_ROTATE);
 
-  pthread_mutex_lock(&LOCK_index);
+  if (need_mutex)
+    pthread_mutex_lock(&LOCK_index);
   if ((error=find_log_pos(&log_info, to_log, 0 /*no mutex*/)))
     goto err;
 
   /*
-    File name exists in index file; Delete until we find this file
+    File name exists in index file; delete until we find this file
     or a file that is used.
   */
   if ((error=find_log_pos(&log_info, NullS, 0 /*no mutex*/)))
     goto err;
-  while (strcmp(to_log,log_info.log_file_name) &&
-	 !log_in_use(log_info.log_file_name))
+  while ((strcmp(to_log,log_info.log_file_name) || (exit_loop=included)) &&
+         !log_in_use(log_info.log_file_name))
   {
-    /* It's not fatal even if we can't delete a log file */
-    my_delete(log_info.log_file_name, MYF(0));
-    if (find_next_log(&log_info, 0))
+    ulong tmp;
+    if (decrease_log_space) //stat the file we want to delete
+    {
+      MY_STAT s;
+      if (my_stat(log_info.log_file_name,&s,MYF(0)))
+        tmp= s.st_size;
+      else
+      {
+        /* 
+           If we could not stat, we can't know the amount
+           of space that deletion will free. In most cases,
+           deletion won't work either, so it's not a problem.
+        */
+        tmp= 0; 
+      }
+    }
+    /*
+      It's not fatal if we can't delete a log file ;
+      if we could delete it, take its size into account
+    */
+    DBUG_PRINT("info",("purging %s",log_info.log_file_name));
+    if (!my_delete(log_info.log_file_name, MYF(0)) && decrease_log_space)
+      *decrease_log_space-= tmp;
+    if (find_next_log(&log_info, 0) || exit_loop)
       break;
   }
 
@@ -754,10 +797,11 @@ int MYSQL_LOG::purge_logs(THD* thd, const char* to_log)
     If we get killed -9 here, the sysadmin would have to edit
     the log index file after restart - otherwise, this should be safe
   */
-  error= update_log_index(&log_info);
+  error= update_log_index(&log_info, need_update_threads);
 
 err:
-  pthread_mutex_unlock(&LOCK_index);
+  if (need_mutex)
+    pthread_mutex_unlock(&LOCK_index);
   DBUG_RETURN(error);
 }
 
@@ -779,7 +823,7 @@ err:
     LOG_INFO_PURGE_NO_ROTATE	Binary file that can't be rotated
 */
 
-int MYSQL_LOG::purge_logs_before_date(THD* thd, time_t purge_time)
+int MYSQL_LOG::purge_logs_before_date(time_t purge_time)
 {
   int error;
   LOG_INFO log_info;
@@ -816,7 +860,7 @@ int MYSQL_LOG::purge_logs_before_date(THD* thd, time_t purge_time)
     If we get killed -9 here, the sysadmin would have to edit
     the log index file after restart - otherwise, this should be safe
   */
-  error= update_log_index(&log_info);
+  error= update_log_index(&log_info, 1);
 
 err:
   pthread_mutex_unlock(&LOCK_index);
@@ -1269,7 +1313,7 @@ err:
   {
     long purge_time= time(0) - expire_logs_days*24*60*60;
     if (purge_time >= 0)
-      error= purge_logs_before_date(current_thd, purge_time);
+      error= purge_logs_before_date(purge_time);
   }
 
 #endif
@@ -1534,7 +1578,6 @@ bool MYSQL_LOG::write(THD *thd,const char *query, uint query_length,
     If you don't do it this way, you will get a deadlock in THD::awake()
 */
 
-
 void MYSQL_LOG:: wait_for_update(THD* thd)
 {
   safe_mutex_assert_owner(&LOCK_log);
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d4efb65bf42..39db264d898 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -310,11 +310,36 @@ int Log_event::exec_event(struct st_relay_log_info* rli)
   */
   if (rli)  
   {
-    if (rli->inside_transaction)
-      rli->inc_pending(get_event_len());
+    /*
+      If in a transaction, and if the slave supports transactions,
+      just inc_event_relay_log_pos(). We only have to check for OPTION_BEGIN
+      (not OPTION_NOT_AUTOCOMMIT) as transactions are logged
+      with BEGIN/COMMIT, not with SET AUTOCOMMIT= .
+      
+      CAUTION: opt_using_transactions means
+      innodb || bdb ; suppose the master supports InnoDB and BDB, 
+      but the slave supports only BDB, problems
+      will arise: 
+      - suppose an InnoDB table is created on the master,
+      - then it will be MyISAM on the slave
+      - but as opt_using_transactions is true, the slave will believe he is
+      transactional with the MyISAM table. And problems will come when one
+      does START SLAVE; STOP SLAVE; START SLAVE; (the slave will resume at BEGIN
+      whereas there has not been any rollback). This is the problem of
+      using opt_using_transactions instead of a finer
+      "does the slave support _the_transactional_handler_used_on_the_master_".
+      
+      More generally, we'll have problems when a query mixes a transactional
+      handler and MyISAM and STOP SLAVE is issued in the middle of the
+      "transaction". START SLAVE will resume at BEGIN while the MyISAM table has
+      already been updated.
+      
+    */
+    if ((thd->options & OPTION_BEGIN) && opt_using_transactions)
+      rli->inc_event_relay_log_pos(get_event_len());
     else
     {
-      rli->inc_pos(get_event_len(),log_pos);
+      rli->inc_group_relay_log_pos(get_event_len(),log_pos);
       flush_relay_log_info(rli);
     }
   }
@@ -878,9 +903,13 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
   thd->db = rewrite_db((char*)db);
 
   /*
-    InnoDB internally stores the master log position it has processed so far;
-    position to store is really pos + pending + event_len
-    since we must store the pos of the END of the current log event
+    InnoDB internally stores the master log position it has executed so far,
+    i.e. the position just after the COMMIT event.
+    When InnoDB will want to store, the positions in rli won't have
+    been updated yet, so group_master_log_* will point to old BEGIN
+    and event_master_log* will point to the beginning of current COMMIT.
+    So the position to store is event_master_log_pos + event_len
+    since we must store the pos of the END of the current log event (COMMIT).
   */
   rli->event_len= get_event_len();
 
@@ -909,18 +938,6 @@ int Query_log_event::exec_event(struct st_relay_log_info* rli)
       DBUG_PRINT("query",("%s",thd->query));
       mysql_parse(thd, thd->query, q_len);
 
-      /*
-	Set a flag if we are inside an transaction so that we can restart
-	the transaction from the start if we are killed
-
-	This will only be done if we are supporting transactional tables
-	in the slave.
-      */
-      if (!strcmp(thd->query,"BEGIN"))
-	rli->inside_transaction= opt_using_transactions;
-      else if (!strcmp(thd->query,"COMMIT"))
-	rli->inside_transaction=0;
-
       DBUG_PRINT("info",("expected_error: %d  last_errno: %d",
 			 expected_error, thd->net.last_errno));
       if ((expected_error != (actual_error= thd->net.last_errno)) &&
@@ -1776,14 +1793,15 @@ int Rotate_log_event::write_data(IO_CACHE* file)
 #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 int Rotate_log_event::exec_event(struct st_relay_log_info* rli)
 {
-  char* log_name = rli->master_log_name;
+  char* log_name = rli->group_master_log_name;
   DBUG_ENTER("Rotate_log_event::exec_event");
 
   pthread_mutex_lock(&rli->data_lock);
   memcpy(log_name, new_log_ident, ident_len+1);
-  rli->master_log_pos = pos;
-  rli->relay_log_pos += get_event_len();
-  DBUG_PRINT("info", ("master_log_pos: %d", (ulong) rli->master_log_pos));
+  rli->group_master_log_pos = pos;
+  rli->event_relay_log_pos += get_event_len();
+  rli->group_relay_log_pos = rli->event_relay_log_pos;
+  DBUG_PRINT("info", ("group_master_log_pos: %d", (ulong) rli->group_master_log_pos));
   pthread_mutex_unlock(&rli->data_lock);
   pthread_cond_broadcast(&rli->data_cond);
   flush_relay_log_info(rli);
@@ -1905,7 +1923,7 @@ int Intvar_log_event::exec_event(struct st_relay_log_info* rli)
     thd->next_insert_id = val;
     break;
   }
-  rli->inc_pending(get_event_len());
+  rli->inc_event_relay_log_pos(get_event_len());
   return 0;
 }
 #endif
@@ -1967,7 +1985,7 @@ int Rand_log_event::exec_event(struct st_relay_log_info* rli)
 {
   thd->rand.seed1= (ulong) seed1;
   thd->rand.seed2= (ulong) seed2;
-  rli->inc_pending(get_event_len());
+  rli->inc_event_relay_log_pos(get_event_len());
   return 0;
 }
 #endif // !MYSQL_CLIENT
@@ -2199,7 +2217,7 @@ int User_var_log_event::exec_event(struct st_relay_log_info* rli)
   e.update_hash(val, val_len, type, charset, Item::COER_NOCOLL);
   free_root(&thd->mem_root,0);
 
-  rli->inc_pending(get_event_len());
+  rli->inc_event_relay_log_pos(get_event_len());
   return 0;
 }
 #endif // !MYSQL_CLIENT
@@ -2241,7 +2259,7 @@ Slave_log_event::Slave_log_event(THD* thd_arg,
   pthread_mutex_lock(&mi->data_lock);
   pthread_mutex_lock(&rli->data_lock);
   master_host_len = strlen(mi->host);
-  master_log_len = strlen(rli->master_log_name);
+  master_log_len = strlen(rli->group_master_log_name);
   // on OOM, just do not initialize the structure and print the error
   if ((mem_pool = (char*)my_malloc(get_data_size() + 1,
 				   MYF(MY_WME))))
@@ -2249,9 +2267,9 @@ Slave_log_event::Slave_log_event(THD* thd_arg,
     master_host = mem_pool + SL_MASTER_HOST_OFFSET ;
     memcpy(master_host, mi->host, master_host_len + 1);
     master_log = master_host + master_host_len + 1;
-    memcpy(master_log, rli->master_log_name, master_log_len + 1);
+    memcpy(master_log, rli->group_master_log_name, master_log_len + 1);
     master_port = mi->port;
-    master_pos = rli->master_log_pos;
+    master_pos = rli->group_master_log_pos;
     DBUG_PRINT("info", ("master_log: %s  pos: %d", master_log,
 			(ulong) master_pos));
   }
@@ -2381,19 +2399,19 @@ void Stop_log_event::print(FILE* file, bool short_form, char* last_db)
 int Stop_log_event::exec_event(struct st_relay_log_info* rli)
 {
   // do not clean up immediately after rotate event
-  if (rli->master_log_pos > BIN_LOG_HEADER_SIZE) 
+  if (rli->group_master_log_pos > BIN_LOG_HEADER_SIZE) 
   {
     close_temporary_tables(thd);
     cleanup_load_tmpdir();
   }
   /*
     We do not want to update master_log pos because we get a rotate event
-    before stop, so by now master_log_name is set to the next log.
+    before stop, so by now group_master_log_name is set to the next log.
     If we updated it, we will have incorrect master coordinates and this
     could give false triggers in MASTER_POS_WAIT() that we have reached
     the target position when in fact we have not.
   */
-  rli->inc_pos(get_event_len(), 0);  
+  rli->inc_group_relay_log_pos(get_event_len(), 0);  
   flush_relay_log_info(rli);
   return 0;
 }
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index 4203d440667..d17faa3cea5 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -721,7 +721,7 @@ extern ulong max_binlog_size, rpl_recovery_rank, thread_cache_size;
 extern ulong com_stat[(uint) SQLCOM_END], com_other, back_log;
 extern ulong specialflag, current_pid;
 extern ulong expire_logs_days;
-
+extern my_bool relay_log_purge;
 extern uint test_flags,select_errors,ha_open_options;
 extern uint protocol_version,dropping_tables;
 extern uint delay_key_write_options;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 47c9c3f8331..6907da7d8e1 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -314,7 +314,7 @@ my_bool opt_safe_user_create = 0, opt_no_mix_types = 0;
 my_bool lower_case_table_names, opt_old_rpl_compat;
 my_bool opt_show_slave_auth_info, opt_sql_bin_update = 0;
 my_bool opt_log_slave_updates= 0, opt_old_passwords=0, use_old_passwords=0;
-my_bool	opt_console= 0;
+my_bool	opt_console= 0, opt_bdb, opt_innodb;
 
 volatile bool  mqh_used = 0;
 FILE *bootstrap_file=0;
@@ -442,6 +442,7 @@ const char **errmesg;			/* Error messages */
 const char *myisam_recover_options_str="OFF";
 const char *sql_mode_str="OFF";
 ulong rpl_recovery_rank=0;
+my_bool relay_log_purge=1;
 
 my_string mysql_unix_port=NULL, opt_mysql_tmpdir=NULL;
 MY_TMPDIR mysql_tmpdir_list;
@@ -2092,9 +2093,9 @@ static int init_common_variables(const char *conf_file_name, int argc,
 #endif
   if (!(default_charset_info= get_charset_by_name(sys_charset.value, MYF(MY_WME))))
     return 1;
-  global_system_variables.result_collation= default_charset_info;
-  global_system_variables.client_collation= default_charset_info;
-  global_system_variables.literal_collation= default_charset_info;
+  global_system_variables.collation_results= default_charset_info;
+  global_system_variables.collation_client= default_charset_info;
+  global_system_variables.collation_connection= default_charset_info;
 
   charsets_list= list_charsets(MYF(MY_CS_COMPILED | MY_CS_CONFIG));
 
@@ -2208,7 +2209,7 @@ static int init_server_components()
     {
       long purge_time= time(0) - expire_logs_days*24*60*60;
       if (purge_time >= 0)
-	mysql_bin_log.purge_logs_before_date(current_thd, purge_time);
+	mysql_bin_log.purge_logs_before_date(purge_time);
     }
 #endif
   }
@@ -3400,7 +3401,7 @@ enum options
   OPT_DELAY_KEY_WRITE,	       OPT_CHARSETS_DIR,
   OPT_BDB_HOME,                OPT_BDB_LOG,
   OPT_BDB_TMP,                 OPT_BDB_NOSYNC,
-  OPT_BDB_LOCK,                OPT_BDB_SKIP,
+  OPT_BDB_LOCK,                OPT_BDB,
   OPT_BDB_NO_RECOVER,	    OPT_BDB_SHARED,
   OPT_MASTER_HOST,             OPT_MASTER_USER,
   OPT_MASTER_PASSWORD,         OPT_MASTER_PORT,
@@ -3430,7 +3431,7 @@ enum options
   OPT_INNODB_FLUSH_METHOD,
   OPT_INNODB_FAST_SHUTDOWN,
   OPT_SAFE_SHOW_DB,
-  OPT_INNODB_SKIP, OPT_SKIP_SAFEMALLOC,
+  OPT_INNODB, OPT_SKIP_SAFEMALLOC,
   OPT_TEMP_POOL, OPT_TX_ISOLATION,
   OPT_SKIP_STACK_TRACE, OPT_SKIP_SYMLINKS,
   OPT_MAX_BINLOG_DUMP_EVENTS, OPT_SPORADIC_BINLOG_DUMP_FAIL,
@@ -3457,7 +3458,7 @@ enum options
   OPT_MAX_BINLOG_CACHE_SIZE, OPT_MAX_BINLOG_SIZE,
   OPT_MAX_CONNECTIONS, OPT_MAX_CONNECT_ERRORS,
   OPT_MAX_DELAYED_THREADS, OPT_MAX_HEP_TABLE_SIZE,
-  OPT_MAX_JOIN_SIZE, OPT_MAX_SORT_LENGTH,
+  OPT_MAX_JOIN_SIZE, OPT_MAX_LENGTH_FOR_SORT_DATA, OPT_MAX_SORT_LENGTH,
   OPT_MAX_TMP_TABLES, OPT_MAX_USER_CONNECTIONS,
   OPT_MAX_WRITE_LOCK_COUNT, OPT_BULK_INSERT_BUFFER_SIZE,
   OPT_MAX_ERROR_COUNT, OPT_MAX_PREP_STMT,
@@ -3468,7 +3469,7 @@ enum options
   OPT_OPEN_FILES_LIMIT,
   OPT_QUERY_CACHE_LIMIT, OPT_QUERY_CACHE_MIN_RES_UNIT, OPT_QUERY_CACHE_SIZE,
   OPT_QUERY_CACHE_TYPE, OPT_RECORD_BUFFER,
-  OPT_RECORD_RND_BUFFER, OPT_RELAY_LOG_SPACE_LIMIT,
+  OPT_RECORD_RND_BUFFER, OPT_RELAY_LOG_SPACE_LIMIT, OPT_RELAY_LOG_PURGE,
   OPT_SLAVE_NET_TIMEOUT, OPT_SLAVE_COMPRESSED_PROTOCOL, OPT_SLOW_LAUNCH_TIME,
   OPT_SORT_BUFFER, OPT_TABLE_CACHE,
   OPT_THREAD_CONCURRENCY, OPT_THREAD_CACHE_SIZE,
@@ -3529,8 +3530,10 @@ struct my_option my_long_options[] =
    (gptr*) &berkeley_tmpdir, (gptr*) &berkeley_tmpdir, 0, GET_STR,
    REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
 #endif /* HAVE_BERKELEY_DB */
-  {"skip-bdb", OPT_BDB_SKIP, "Don't use berkeley db (will save memory)",
-   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"bdb", OPT_BDB, "Enable Berkeley DB (if this version of MySQL supports it). \
+Disable with --skip-bdb (will save memory)",
+   (gptr*) &opt_bdb, (gptr*) &opt_bdb, 0, GET_BOOL, NO_ARG, 1, 0, 0,
+   0, 0, 0},
   {"big-tables", OPT_BIG_TABLES,
    "Allow big result sets by saving all temporary sets on file (Solves most 'table full' errors)",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
@@ -3885,8 +3888,10 @@ struct my_option my_long_options[] =
    "Start without grant tables. This gives all users FULL ACCESS to all tables!",
    (gptr*) &opt_noacl, (gptr*) &opt_noacl, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0,
    0},
-  {"skip-innodb", OPT_INNODB_SKIP, "Don't use Innodb (will save memory)",
-   0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"innodb", OPT_INNODB, "Enable InnoDB (if this version of MySQL supports it). \
+Disable with --skip-innodb (will save memory)",
+   (gptr*) &opt_innodb, (gptr*) &opt_innodb, 0, GET_BOOL, NO_ARG, 1, 0, 0,
+   0, 0, 0},
   {"skip-locking", OPT_SKIP_LOCK,
    "Deprecated option, use --skip-external-locking instead",
    0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0},
@@ -4159,6 +4164,11 @@ struct my_option my_long_options[] =
    (gptr*) &global_system_variables.max_join_size,
    (gptr*) &max_system_variables.max_join_size, 0, GET_HA_ROWS, REQUIRED_ARG,
    ~0L, 1, ~0L, 0, 1, 0},
+   {"max_length_for_sort_data", OPT_MAX_LENGTH_FOR_SORT_DATA,
+    "Max number of bytes in sorted records",
+    (gptr*) &global_system_variables.max_length_for_sort_data,
+    (gptr*) &max_system_variables.max_length_for_sort_data, 0, GET_ULONG,
+    REQUIRED_ARG, 1024, 4, 8192*1024L, 0, 1, 0},
   {"max_prepared_statements", OPT_MAX_PREP_STMT,
    "Max number of prepared_statements for a thread",
    (gptr*) &global_system_variables.max_prep_stmt_count,
@@ -4272,6 +4282,11 @@ struct my_option my_long_options[] =
    (gptr*) &max_system_variables.read_buff_size,0, GET_ULONG, REQUIRED_ARG,
    128*1024L, IO_SIZE*2+MALLOC_OVERHEAD, ~0L, MALLOC_OVERHEAD, IO_SIZE, 0},
 #ifdef HAVE_REPLICATION
+  {"relay_log_purge", OPT_RELAY_LOG_PURGE,
+   "0 = do not purge relay logs. 1 = purge them as soon as they are no more needed.",
+   (gptr*) &relay_log_purge,
+   (gptr*) &relay_log_purge, 0, GET_BOOL, NO_ARG,
+   1, 0, 1, 0, 1, 0},
   {"relay_log_space_limit", OPT_RELAY_LOG_SPACE_LIMIT,
    "Max space to use for all relay logs",
    (gptr*) &relay_log_space_limit,
@@ -4575,9 +4590,9 @@ static void set_options(void)
 		 sizeof(mysql_real_data_home)-1);
 
   /* Set default values for some variables */
-  global_system_variables.result_collation= default_charset_info;
-  global_system_variables.client_collation= default_charset_info;
-  global_system_variables.literal_collation= default_charset_info;
+  global_system_variables.collation_results= default_charset_info;
+  global_system_variables.collation_client= default_charset_info;
+  global_system_variables.collation_connection= default_charset_info;
   global_system_variables.table_type=   DB_TYPE_MYISAM;
   global_system_variables.tx_isolation= ISO_REPEATABLE_READ;
   global_system_variables.select_limit= (ulonglong) HA_POS_ERROR;
@@ -5016,16 +5031,32 @@ get_one_option(int optid, const struct my_option *opt __attribute__((unused)),
     berkeley_shared_data=1;
     break;
 #endif /* HAVE_BERKELEY_DB */
-  case OPT_BDB_SKIP:
+  case OPT_BDB:
 #ifdef HAVE_BERKELEY_DB
-    berkeley_skip=1;
-    have_berkeley_db=SHOW_OPTION_DISABLED;
+    if (opt_bdb)
+    {
+      berkeley_skip=0;
+      have_berkeley_db=SHOW_OPTION_YES;
+    }
+    else
+    {
+      berkeley_skip=1;
+      have_berkeley_db=SHOW_OPTION_DISABLED;
+    }
 #endif
     break;
-  case OPT_INNODB_SKIP:
+  case OPT_INNODB:
 #ifdef HAVE_INNOBASE_DB
-    innodb_skip=1;
-    have_innodb=SHOW_OPTION_DISABLED;
+    if (opt_innodb)
+    {
+      innodb_skip=0;
+      have_innodb=SHOW_OPTION_YES;
+    }
+    else
+    {
+      innodb_skip=1;
+      have_innodb=SHOW_OPTION_DISABLED;
+    }
 #endif
     break;
   case OPT_INNODB_DATA_FILE_PATH:
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 92bab76bedd..7526772cd09 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -349,13 +349,13 @@ SQL_SELECT *make_select(TABLE *head, table_map const_tables,
   select->head=head;
   select->cond=conds;
 
-  if (head->io_cache)
+  if (head->sort.io_cache)
   {
-    select->file= *head->io_cache;
+    select->file= *head->sort.io_cache;
     select->records=(ha_rows) (select->file.end_of_file/
 			       head->file->ref_length);
-    my_free((gptr) (head->io_cache),MYF(0));
-    head->io_cache=0;
+    my_free((gptr) (head->sort.io_cache),MYF(0));
+    head->sort.io_cache=0;
   }
   DBUG_RETURN(select);
 }
diff --git a/sql/protocol.cc b/sql/protocol.cc
index 99b815a7840..d00ecb5dbc4 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -724,7 +724,7 @@ bool Protocol_simple::store(const char *from, uint length,
 bool Protocol_simple::store(const char *from, uint length,
 			    CHARSET_INFO *fromcs)
 {
-  CHARSET_INFO *tocs= this->thd->variables.result_collation;
+  CHARSET_INFO *tocs= this->thd->variables.collation_results;
 #ifndef DEBUG_OFF
   DBUG_ASSERT(field_types == 0 ||
 	      field_types[field_pos] == MYSQL_TYPE_DECIMAL ||
diff --git a/sql/records.cc b/sql/records.cc
index 22c4d54550c..e6c6e62a516 100644
--- a/sql/records.cc
+++ b/sql/records.cc
@@ -1,3 +1,4 @@
+
 /* Copyright (C) 2000 MySQL AB & MySQL Finland AB & TCX DataKonsult AB
 
    This program is free software; you can redistribute it and/or modify
@@ -22,6 +23,8 @@
 static int rr_quick(READ_RECORD *info);
 static int rr_sequential(READ_RECORD *info);
 static int rr_from_tempfile(READ_RECORD *info);
+static int rr_unpack_from_tempfile(READ_RECORD *info);
+static int rr_unpack_from_buffer(READ_RECORD *info);
 static int rr_from_pointers(READ_RECORD *info);
 static int rr_from_cache(READ_RECORD *info);
 static int init_rr_cache(READ_RECORD *info);
@@ -41,8 +44,16 @@ void init_read_record(READ_RECORD *info,THD *thd, TABLE *table,
   info->table=table;
   info->file= table->file;
   info->forms= &info->table;		/* Only one table */
-  info->record=table->record[0];
-  info->ref_length=table->file->ref_length;
+  if (table->sort.addon_field)
+  {
+    info->rec_buf= table->sort.addon_buf;
+    info->ref_length= table->sort.addon_length;
+  }
+  else
+  {
+    info->record= table->record[0];
+    info->ref_length= table->file->ref_length;
+  }
   info->select=select;
   info->print_error=print_error;
   info->ignore_not_found_rows= 0;
@@ -51,11 +62,12 @@ void init_read_record(READ_RECORD *info,THD *thd, TABLE *table,
   if (select && my_b_inited(&select->file))
     tempfile= &select->file;
   else
-    tempfile= table->io_cache;
+    tempfile= table->sort.io_cache;
   if (tempfile && my_b_inited(tempfile)) // Test if ref-records was used
   {
     DBUG_PRINT("info",("using rr_from_tempfile"));
-    info->read_record=rr_from_tempfile;
+    info->read_record= (table->sort.addon_field ?
+                        rr_unpack_from_tempfile : rr_from_tempfile);
     info->io_cache=tempfile;
     reinit_io_cache(info->io_cache,READ_CACHE,0L,0,0);
     info->ref_pos=table->file->ref;
@@ -85,13 +97,15 @@ void init_read_record(READ_RECORD *info,THD *thd, TABLE *table,
     DBUG_PRINT("info",("using rr_quick"));
     info->read_record=rr_quick;
   }
-  else if (table->record_pointers)
+  else if (table->sort.record_pointers)
   {
     DBUG_PRINT("info",("using record_pointers"));
     table->file->rnd_init(0);
-    info->cache_pos=table->record_pointers;
-    info->cache_end=info->cache_pos+ table->found_records*info->ref_length;
-    info->read_record= rr_from_pointers;
+    info->cache_pos=table->sort.record_pointers;
+    info->cache_end=info->cache_pos+ 
+                    table->sort.found_records*info->ref_length;
+    info->read_record= (table->sort.addon_field ?
+                        rr_unpack_from_buffer : rr_from_pointers);
   }
   else
   {
@@ -112,7 +126,7 @@ void init_read_record(READ_RECORD *info,THD *thd, TABLE *table,
 
 
 void end_read_record(READ_RECORD *info)
-{					/* free cache if used */
+{                   /* free cache if used */
   if (info->cache)
   {
     my_free_lock((char*) info->cache,MYF(0));
@@ -120,6 +134,19 @@ void end_read_record(READ_RECORD *info)
   }
   if (info->table)
   {
+    TABLE *table= info->table;
+    if (table->sort.record_pointers)
+    {
+      my_free((gptr) table->sort.record_pointers,MYF(0));
+      table->sort.record_pointers=0;
+    }
+    if (table->sort.addon_buf)
+    {
+      my_free((char *) table->sort.addon_buf, MYF(0));
+      my_free((char *) table->sort.addon_field, MYF(MY_ALLOW_ZERO_PTR));
+      table->sort.addon_buf=0;
+      table->sort.addon_field=0;
+    }
     (void) info->file->extra(HA_EXTRA_NO_CACHE);
     (void) info->file->rnd_end();
     info->table=0;
@@ -200,6 +227,34 @@ tryNext:
 } /* rr_from_tempfile */
 
 
+/*
+  Read a result set record from a temporary file after sorting
+
+  SYNOPSIS
+    rr_unpack_from_tempfile()
+    info                Reference to the context including record descriptors
+
+  DESCRIPTION
+    The function first reads the next sorted record from the temporary file.
+    into a buffer. If a success it calls a callback function that unpacks 
+    the fields values use in the result set from this buffer into their
+    positions in the regular record buffer.
+
+  RETURN
+     0 - Record successfully read.
+    -1 - There is no record to be read anymore. 
+*/
+
+static int rr_unpack_from_tempfile(READ_RECORD *info)
+{
+  if (my_b_read(info->io_cache, info->rec_buf, info->ref_length))
+    return -1;
+  TABLE *table= info->table;
+  (*table->sort.unpack)(table->sort.addon_field, info->rec_buf);
+
+  return 0;
+}
+
 static int rr_from_pointers(READ_RECORD *info)
 {
   int tmp;
@@ -228,6 +283,34 @@ tryNext:
   return tmp;
 }
 
+/*
+  Read a result set record from a buffer after sorting
+
+  SYNOPSIS
+    rr_unpack_from_buffer()
+    info                   Reference to the context including record descriptors
+
+  DESCRIPTION
+    The function first reads the next sorted record from the sort buffer.
+    If a success it calls a callback function that unpacks 
+    the fields values use in the result set from this buffer into their
+    positions in the regular record buffer.
+
+  RETURN
+     0 - Record successfully read.
+    -1 - There is no record to be read anymore. 
+*/
+
+static int rr_unpack_from_buffer(READ_RECORD *info)
+{
+  if (info->cache_pos == info->cache_end)
+    return -1;                      /* End of buffer */
+  TABLE *table= info->table;
+  (*table->sort.unpack)(table->sort.addon_field, info->cache_pos);
+  info->cache_pos+= info->ref_length;
+
+  return 0;
+}
 	/* cacheing of records from a database */
 
 static int init_rr_cache(READ_RECORD *info)
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index 46791c13219..58769827bed 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -873,7 +873,6 @@ int load_master_data(THD* thd)
 	// don't hit the magic number
 	if (active_mi->master_log_pos < BIN_LOG_HEADER_SIZE)
 	  active_mi->master_log_pos = BIN_LOG_HEADER_SIZE;
-	active_mi->rli.pending = 0;
 	flush_master_info(active_mi);
       }
       mc_mysql_free_result(master_status_res);
@@ -897,9 +896,13 @@ int load_master_data(THD* thd)
     return 1;
   }
   pthread_mutex_lock(&active_mi->rli.data_lock);
-  active_mi->rli.master_log_pos = active_mi->master_log_pos;
-  strmake(active_mi->rli.master_log_name,active_mi->master_log_name,
-	  sizeof(active_mi->rli.master_log_name)-1);
+  active_mi->rli.group_master_log_pos = active_mi->master_log_pos;
+  strmake(active_mi->rli.group_master_log_name,active_mi->master_log_name,
+	  sizeof(active_mi->rli.group_master_log_name)-1);
+  /*
+    No need to update rli.event* coordinates, they will be when the slave
+    threads start ; only rli.group* coordinates are necessary here.
+  */
   flush_relay_log_info(&active_mi->rli);
   pthread_cond_broadcast(&active_mi->rli.data_cond);
   pthread_mutex_unlock(&active_mi->rli.data_lock);
diff --git a/sql/set_var.cc b/sql/set_var.cc
index d03b91ef83b..e4adbb0a318 100644
--- a/sql/set_var.cc
+++ b/sql/set_var.cc
@@ -105,7 +105,9 @@ sys_var_str		sys_charset("character_set",
 				    sys_check_charset,
 				    sys_update_charset,
 				    sys_set_default_charset);
-sys_var_client_collation sys_client_collation("client_collation");
+sys_var_collation_client sys_collation_client("collation_client");
+sys_var_collation_connection sys_collation_connection("collation_connection");
+sys_var_collation_results sys_collation_results("collation_results");
 sys_var_bool_ptr	sys_concurrent_insert("concurrent_insert",
 					      &myisam_concurrent_insert);
 sys_var_long_ptr	sys_connect_timeout("connect_timeout",
@@ -131,7 +133,6 @@ sys_var_thd_ulong	sys_join_buffer_size("join_buffer_size",
 sys_var_ulonglong_ptr	sys_key_buffer_size("key_buffer_size",
 					    &keybuff_size,
 					    fix_key_buffer_size);
-sys_var_literal_collation sys_literal_collation("literal_collation");
 sys_var_bool_ptr	sys_local_infile("local_infile",
 					 &opt_local_infile);
 sys_var_thd_bool	sys_log_warnings("log_warnings", &SV::log_warnings);
@@ -166,6 +167,8 @@ sys_var_thd_ulong       sys_pseudo_thread_id("pseudo_thread_id",
 sys_var_thd_ha_rows	sys_max_join_size("max_join_size",
 					  &SV::max_join_size,
 					  fix_max_join_size);
+sys_var_thd_ulong   sys_max_length_for_sort_data("max_length_for_sort_data",
+                        &SV::max_length_for_sort_data);
 #ifndef TO_BE_DELETED	/* Alias for max_join_size */
 sys_var_thd_ha_rows	sys_sql_max_join_size("sql_max_join_size",
 					      &SV::max_join_size,
@@ -200,7 +203,10 @@ sys_var_thd_ulong	sys_read_buff_size("read_buffer_size",
 					   &SV::read_buff_size);
 sys_var_thd_ulong	sys_read_rnd_buff_size("read_rnd_buffer_size",
 					       &SV::read_rnd_buff_size);
-sys_var_result_collation sys_result_collation("result_collation");
+#ifdef HAVE_REPLICATION
+sys_var_bool_ptr	sys_relay_log_purge("relay_log_purge",
+                                            &relay_log_purge);
+#endif
 sys_var_long_ptr	sys_rpl_recovery_rank("rpl_recovery_rank",
 					      &rpl_recovery_rank);
 sys_var_long_ptr	sys_query_cache_size("query_cache_size",
@@ -343,7 +349,9 @@ sys_var *sys_variables[]=
   &sys_binlog_cache_size,
   &sys_buffer_results,
   &sys_bulk_insert_buff_size,
-  &sys_client_collation,
+  &sys_collation_client,
+  &sys_collation_connection,
+  &sys_collation_results,
   &sys_concurrent_insert,
   &sys_connect_timeout,
   &sys_default_week_format,
@@ -362,7 +370,6 @@ sys_var *sys_variables[]=
   &sys_interactive_timeout,
   &sys_join_buffer_size,
   &sys_key_buffer_size,
-  &sys_literal_collation,
   &sys_last_insert_id,
   &sys_local_infile,
   &sys_log_binlog,
@@ -380,6 +387,7 @@ sys_var *sys_variables[]=
   &sys_max_error_count,
   &sys_max_heap_table_size,
   &sys_max_join_size,
+  &sys_max_length_for_sort_data,
   &sys_max_prep_stmt_count,
   &sys_max_sort_length,
   &sys_max_tmp_tables,
@@ -406,7 +414,9 @@ sys_var *sys_variables[]=
   &sys_rand_seed2,
   &sys_read_buff_size,
   &sys_read_rnd_buff_size,
-  &sys_result_collation,
+#ifdef HAVE_REPLICATION
+  &sys_relay_log_purge,
+#endif
   &sys_rpl_recovery_rank,
   &sys_safe_updates,
   &sys_select_limit,
@@ -455,7 +465,9 @@ struct show_var_st init_vars[]= {
   {sys_bulk_insert_buff_size.name,(char*) &sys_bulk_insert_buff_size,SHOW_SYS},
   {sys_charset.name, 	      (char*) &sys_charset,		     SHOW_SYS},
   {"character_sets",          (char*) &charsets_list,               SHOW_CHAR_PTR},
-  {sys_client_collation.name, (char*) &sys_client_collation,	    SHOW_SYS},
+  {sys_collation_client.name, (char*) &sys_collation_client,	    SHOW_SYS},
+  {sys_collation_connection.name,(char*) &sys_collation_connection, SHOW_SYS},
+  {sys_collation_results.name, (char*) &sys_collation_results,	    SHOW_SYS},
   {sys_concurrent_insert.name,(char*) &sys_concurrent_insert,       SHOW_SYS},
   {sys_connect_timeout.name,  (char*) &sys_connect_timeout,         SHOW_SYS},
   {"datadir",                 mysql_real_data_home,                 SHOW_CHAR},
@@ -508,7 +520,6 @@ struct show_var_st init_vars[]= {
   {sys_key_buffer_size.name,	(char*) &sys_key_buffer_size,	    SHOW_SYS},
   {"language",                language,                             SHOW_CHAR},
   {"large_files_support",     (char*) &opt_large_files,             SHOW_BOOL},	
-  {sys_literal_collation.name,(char*) &sys_literal_collation,	    SHOW_SYS},
   {sys_local_infile.name,     (char*) &sys_local_infile,	    SHOW_SYS},
 #ifdef HAVE_MLOCKALL
   {"locked_in_memory",	      (char*) &locked_in_memory,	    SHOW_BOOL},
@@ -533,6 +544,9 @@ struct show_var_st init_vars[]= {
   {sys_max_delayed_threads.name,(char*) &sys_max_delayed_threads,   SHOW_SYS},
   {sys_max_heap_table_size.name,(char*) &sys_max_heap_table_size,   SHOW_SYS},
   {sys_max_join_size.name,	(char*) &sys_max_join_size,	    SHOW_SYS},
+  {sys_max_length_for_sort_data.name,
+   (char*) &sys_max_length_for_sort_data,
+   SHOW_SYS},
   {sys_max_prep_stmt_count.name,(char*) &sys_max_prep_stmt_count,   SHOW_SYS},
   {sys_max_sort_length.name,	(char*) &sys_max_sort_length,	    SHOW_SYS},
   {sys_max_user_connections.name,(char*) &sys_max_user_connections, SHOW_SYS},
@@ -562,7 +576,9 @@ struct show_var_st init_vars[]= {
   {sys_pseudo_thread_id.name, (char*) &sys_pseudo_thread_id,        SHOW_SYS},
   {sys_read_buff_size.name,   (char*) &sys_read_buff_size,	    SHOW_SYS},
   {sys_read_rnd_buff_size.name,(char*) &sys_read_rnd_buff_size,	    SHOW_SYS},
-  {sys_result_collation.name, (char*) &sys_result_collation,	    SHOW_SYS},
+#ifdef HAVE_REPLICATION
+  {sys_relay_log_purge.name,  (char*) &sys_relay_log_purge,         SHOW_SYS},
+#endif
   {sys_rpl_recovery_rank.name,(char*) &sys_rpl_recovery_rank,       SHOW_SYS},
 #ifdef HAVE_QUERY_CACHE
   {sys_query_cache_limit.name,(char*) &sys_query_cache_limit,	    SHOW_SYS},
@@ -1210,86 +1226,80 @@ bool sys_var_collation::check(THD *thd, set_var *var)
   return 0;
 }
 
-bool sys_var_client_collation::update(THD *thd, set_var *var)
+bool sys_var_collation_client::update(THD *thd, set_var *var)
 {
   if (var->type == OPT_GLOBAL)
-    global_system_variables.client_collation= var->save_result.charset;
+    global_system_variables.collation_client= var->save_result.charset;
   else
-  {
-    thd->variables.client_collation= var->save_result.charset;
-    thd->protocol_simple.init(thd);
-    thd->protocol_prep.init(thd);
-  }
+    thd->variables.collation_client= var->save_result.charset;
   return 0;
 }
 
-byte *sys_var_client_collation::value_ptr(THD *thd, enum_var_type type)
+byte *sys_var_collation_client::value_ptr(THD *thd, enum_var_type type)
 {
   CHARSET_INFO *cs= ((type == OPT_GLOBAL) ?
-		  global_system_variables.client_collation :
-		  thd->variables.client_collation);
+		  global_system_variables.collation_client :
+		  thd->variables.collation_client);
   return cs ? (byte*) cs->name : (byte*) "";
 }
 
-void sys_var_client_collation::set_default(THD *thd, enum_var_type type)
+void sys_var_collation_client::set_default(THD *thd, enum_var_type type)
 {
  if (type == OPT_GLOBAL)
-   global_system_variables.client_collation= default_charset_info;
+   global_system_variables.collation_client= default_charset_info;
  else
- {
-   thd->variables.client_collation= global_system_variables.client_collation;
- }
+   thd->variables.collation_client= global_system_variables.collation_client;
 }
 
 
-bool sys_var_literal_collation::update(THD *thd, set_var *var)
+bool sys_var_collation_connection::update(THD *thd, set_var *var)
 {
   if (var->type == OPT_GLOBAL)
-    global_system_variables.literal_collation= var->save_result.charset;
+    global_system_variables.collation_connection= var->save_result.charset;
   else
-    thd->variables.literal_collation= var->save_result.charset;
+    thd->variables.collation_connection= var->save_result.charset;
   return 0;
 }
 
-byte *sys_var_literal_collation::value_ptr(THD *thd, enum_var_type type)
+byte *sys_var_collation_connection::value_ptr(THD *thd, enum_var_type type)
 {
   CHARSET_INFO *cs= ((type == OPT_GLOBAL) ?
-		  global_system_variables.literal_collation :
-		  thd->variables.literal_collation);
+		  global_system_variables.collation_connection :
+		  thd->variables.collation_connection);
   return cs ? (byte*) cs->name : (byte*) "";
 }
 
-void sys_var_literal_collation::set_default(THD *thd, enum_var_type type)
+void sys_var_collation_connection::set_default(THD *thd, enum_var_type type)
 {
  if (type == OPT_GLOBAL)
-   global_system_variables.literal_collation= default_charset_info;
+   global_system_variables.collation_connection= default_charset_info;
  else
-   thd->variables.literal_collation= global_system_variables.literal_collation;
+   thd->variables.collation_connection= global_system_variables.collation_connection;
 }
 
-bool sys_var_result_collation::update(THD *thd, set_var *var)
+bool sys_var_collation_results::update(THD *thd, set_var *var)
 {
   if (var->type == OPT_GLOBAL)
-    global_system_variables.result_collation= var->save_result.charset;
+    global_system_variables.collation_results= var->save_result.charset;
   else
-    thd->variables.result_collation= var->save_result.charset;
+    thd->variables.collation_results= var->save_result.charset;
   return 0;
 }
 
-byte *sys_var_result_collation::value_ptr(THD *thd, enum_var_type type)
+byte *sys_var_collation_results::value_ptr(THD *thd, enum_var_type type)
 {
   CHARSET_INFO *cs= ((type == OPT_GLOBAL) ?
-		  global_system_variables.result_collation :
-		  thd->variables.result_collation);
+		  global_system_variables.collation_results :
+		  thd->variables.collation_results);
   return cs ? (byte*) cs->name : (byte*) "";
 }
 
-void sys_var_result_collation::set_default(THD *thd, enum_var_type type)
+void sys_var_collation_results::set_default(THD *thd, enum_var_type type)
 {
  if (type == OPT_GLOBAL)
-   global_system_variables.result_collation= default_charset_info;
+   global_system_variables.collation_results= default_charset_info;
  else
-   thd->variables.result_collation= global_system_variables.result_collation;
+   thd->variables.collation_results= global_system_variables.collation_results;
 }
 
 
@@ -1297,16 +1307,16 @@ void sys_var_result_collation::set_default(THD *thd, enum_var_type type)
   Functions to handle SET NAMES and SET CHARACTER SET
 *****************************************************************************/
 
-int set_var_client_collation::check(THD *thd)
+int set_var_collation_client::check(THD *thd)
 {
   return 0;
 }
 
-int set_var_client_collation::update(THD *thd)
+int set_var_collation_client::update(THD *thd)
 {
-  thd->variables.client_collation=  client_collation;
-  thd->variables.literal_collation= literal_collation;
-  thd->variables.result_collation=  result_collation;
+  thd->variables.collation_client= collation_client;
+  thd->variables.collation_connection= collation_connection;
+  thd->variables.collation_results= collation_results;
   thd->protocol_simple.init(thd);
   thd->protocol_prep.init(thd);
   return 0;
diff --git a/sql/set_var.h b/sql/set_var.h
index 69e6bc05a67..fbd20228d24 100644
--- a/sql/set_var.h
+++ b/sql/set_var.h
@@ -420,28 +420,28 @@ SHOW_TYPE type() { return SHOW_CHAR; }
   virtual void set_default(THD *thd, enum_var_type type)= 0;
 };
 
-class sys_var_client_collation :public sys_var_collation
+class sys_var_collation_client :public sys_var_collation
 {
 public:
-  sys_var_client_collation(const char *name_arg) :sys_var_collation(name_arg) {}
+  sys_var_collation_client(const char *name_arg) :sys_var_collation(name_arg) {}
   bool update(THD *thd, set_var *var);
   void set_default(THD *thd, enum_var_type type);
   byte *value_ptr(THD *thd, enum_var_type type);
 };
 
-class sys_var_literal_collation :public sys_var_collation
+class sys_var_collation_connection :public sys_var_collation
 {
 public:
-  sys_var_literal_collation(const char *name_arg) :sys_var_collation(name_arg) {}
+  sys_var_collation_connection(const char *name_arg) :sys_var_collation(name_arg) {}
   bool update(THD *thd, set_var *var);
   void set_default(THD *thd, enum_var_type type);
   byte *value_ptr(THD *thd, enum_var_type type);
 };
 
-class sys_var_result_collation :public sys_var_collation
+class sys_var_collation_results :public sys_var_collation
 {
 public:
-  sys_var_result_collation(const char *name_arg) :sys_var_collation(name_arg) {}
+  sys_var_collation_results(const char *name_arg) :sys_var_collation(name_arg) {}
   bool update(THD *thd, set_var *var);
   void set_default(THD *thd, enum_var_type type);
   byte *value_ptr(THD *thd, enum_var_type type);
@@ -553,18 +553,18 @@ public:
 
 /* For SET NAMES and SET CHARACTER SET */
 
-class set_var_client_collation: public set_var_base
+class set_var_collation_client: public set_var_base
 {
-  CHARSET_INFO *client_collation;
-  CHARSET_INFO *literal_collation;
-  CHARSET_INFO *result_collation;
+  CHARSET_INFO *collation_client;
+  CHARSET_INFO *collation_connection;
+  CHARSET_INFO *collation_results;
 public:
-  set_var_client_collation(CHARSET_INFO *client_coll_arg,
-  			   CHARSET_INFO *literal_coll_arg,
+  set_var_collation_client(CHARSET_INFO *client_coll_arg,
+  			   CHARSET_INFO *connection_coll_arg,
   			   CHARSET_INFO *result_coll_arg)
-    :client_collation(client_coll_arg),
-     literal_collation(literal_coll_arg),
-     result_collation(result_coll_arg)
+    :collation_client(client_coll_arg),
+     collation_connection(connection_coll_arg),
+     collation_results(result_coll_arg)
   {}
   int check(THD *thd);
   int update(THD *thd);
diff --git a/sql/slave.cc b/sql/slave.cc
index fa7ab180eaa..eae795ae760 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -217,11 +217,7 @@ static byte* get_table_key(TABLE_RULE_ENT* e, uint* len,
   - If not, open the 'log' binary file.
 
   TODO
-    - check proper initialization of master_log_name/master_log_pos
-    - We may always want to delete all logs before 'log'.
-      Currently if we are not calling this with 'log' as NULL or the first
-      log we will never delete relay logs.
-      If we want this we should not set skip_log_purge to 1.
+    - check proper initialization of group_master_log_name/group_master_log_pos
 
   RETURN VALUES
     0	ok
@@ -248,7 +244,7 @@ int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log,
     rli->cur_log_fd = -1;
   }
   
-  rli->relay_log_pos = pos;
+  rli->group_relay_log_pos = rli->event_relay_log_pos = pos;
 
   /*
     Test to see if the previous run was with the skip of purging
@@ -260,18 +256,15 @@ int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log,
     goto err;
   }
 
-  if (log)					// If not first log
+  if (log && rli->relay_log.find_log_pos(&rli->linfo, log, 1))
   {
-    if (strcmp(log, rli->linfo.log_file_name))
-      rli->skip_log_purge= 1;			// Different name; Don't purge
-    if (rli->relay_log.find_log_pos(&rli->linfo, log, 1))
-    {
-      *errmsg="Could not find target log during relay log initialization";
-      goto err;
-    }
+    *errmsg="Could not find target log during relay log initialization";
+    goto err;
   }
-  strmake(rli->relay_log_name,rli->linfo.log_file_name,
-	  sizeof(rli->relay_log_name)-1);
+  strmake(rli->group_relay_log_name,rli->linfo.log_file_name,
+	  sizeof(rli->group_relay_log_name)-1);
+  strmake(rli->event_relay_log_name,rli->linfo.log_file_name,
+	  sizeof(rli->event_relay_log_name)-1);
   if (rli->relay_log.is_active(rli->linfo.log_file_name))
   {
     /*
@@ -302,7 +295,7 @@ err:
     If we don't purge, we can't honour relay_log_space_limit ;
     silently discard it
   */
-  if (rli->skip_log_purge)
+  if (!relay_log_purge)
     rli->log_space_limit= 0;
   pthread_cond_broadcast(&rli->data_cond);
   if (need_data_lock)
@@ -383,9 +376,8 @@ int purge_relay_logs(RELAY_LOG_INFO* rli, THD *thd, bool just_reset,
     to display fine in any case.
   */
 
-  rli->master_log_name[0]= 0;
-  rli->master_log_pos= 0;
-  rli->pending= 0;
+  rli->group_master_log_name[0]= 0;
+  rli->group_master_log_pos= 0;
 
   if (!rli->inited)
     DBUG_RETURN(0);
@@ -402,16 +394,18 @@ int purge_relay_logs(RELAY_LOG_INFO* rli, THD *thd, bool just_reset,
     goto err;
   }
   /* Save name of used relay log file */
-  strmake(rli->relay_log_name, rli->relay_log.get_log_fname(),
-	  sizeof(rli->relay_log_name)-1);
+  strmake(rli->group_relay_log_name, rli->relay_log.get_log_fname(),
+	  sizeof(rli->group_relay_log_name)-1);
+  strmake(rli->event_relay_log_name, rli->relay_log.get_log_fname(),
+ 	  sizeof(rli->event_relay_log_name)-1);
   // Just first log with magic number and nothing else
   rli->log_space_total= BIN_LOG_HEADER_SIZE;
-  rli->relay_log_pos=   BIN_LOG_HEADER_SIZE;
+  rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE;
   rli->relay_log.reset_bytes_written();
   if (!just_reset)
-    error= init_relay_log_pos(rli, rli->relay_log_name, rli->relay_log_pos,
-			      0 /* do not need data lock */, errmsg);
-
+    error= init_relay_log_pos(rli, rli->group_relay_log_name, rli->group_relay_log_pos,
+  			      0 /* do not need data lock */, errmsg);
+  
 err:
 #ifndef DBUG_OFF
   char buf[22];
@@ -1238,13 +1232,11 @@ int init_relay_log_info(RELAY_LOG_INFO* rli, const char* info_fname)
   fn_format(fname, info_fname, mysql_data_home, "", 4+32);
   pthread_mutex_lock(&rli->data_lock);
   info_fd = rli->info_fd;
-  rli->pending = 0;
   rli->cur_log_fd = -1;
   rli->slave_skip_counter=0;
   rli->abort_pos_wait=0;
-  rli->skip_log_purge=0;
-  rli->log_space_limit = relay_log_space_limit;
-  rli->log_space_total = 0;
+  rli->log_space_limit= relay_log_space_limit;
+  rli->log_space_total= 0;
 
   // TODO: make this work with multi-master
   if (!opt_relay_logname)
@@ -1285,8 +1277,8 @@ int init_relay_log_info(RELAY_LOG_INFO* rli, const char* info_fname)
     if (init_relay_log_pos(rli,NullS,BIN_LOG_HEADER_SIZE,0 /* no data lock */,
 			   &msg))
       goto err;
-    rli->master_log_name[0]= 0;
-    rli->master_log_pos= 0;		
+    rli->group_master_log_name[0]= 0;
+    rli->group_master_log_pos= 0;		
     rli->info_fd= info_fd;
   }
   else // file exists
@@ -1307,31 +1299,33 @@ int init_relay_log_info(RELAY_LOG_INFO* rli, const char* info_fname)
       
     rli->info_fd = info_fd;
     int relay_log_pos, master_log_pos;
-    if (init_strvar_from_file(rli->relay_log_name,
-			      sizeof(rli->relay_log_name), &rli->info_file,
+    if (init_strvar_from_file(rli->group_relay_log_name,
+			      sizeof(rli->group_relay_log_name), &rli->info_file,
 			      "") ||
        init_intvar_from_file(&relay_log_pos,
 			     &rli->info_file, BIN_LOG_HEADER_SIZE) ||
-       init_strvar_from_file(rli->master_log_name,
-			     sizeof(rli->master_log_name), &rli->info_file,
+       init_strvar_from_file(rli->group_master_log_name,
+			     sizeof(rli->group_master_log_name), &rli->info_file,
 			     "") ||
        init_intvar_from_file(&master_log_pos, &rli->info_file, 0))
     {
       msg="Error reading slave log configuration";
       goto err;
     }
-    rli->relay_log_pos=  relay_log_pos;
-    rli->master_log_pos= master_log_pos;
+    strmake(rli->event_relay_log_name,rli->group_relay_log_name,
+            sizeof(rli->event_relay_log_name)-1);
+    rli->group_relay_log_pos= rli->event_relay_log_pos= relay_log_pos;
+    rli->group_master_log_pos= master_log_pos;
 
     if (init_relay_log_pos(rli,
-			   rli->relay_log_name,
-			   rli->relay_log_pos,
+			   rli->group_relay_log_name,
+			   rli->group_relay_log_pos,
 			   0 /* no data lock*/,
 			   &msg))
       goto err;
   }
-  DBUG_ASSERT(rli->relay_log_pos >= BIN_LOG_HEADER_SIZE);
-  DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->relay_log_pos);
+  DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
+  DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
   /*
     Now change the cache from READ to WRITE - must do this
     before flush_relay_log_info
@@ -1407,7 +1401,7 @@ static int count_relay_log_space(RELAY_LOG_INFO* rli)
 {
   LOG_INFO linfo;
   DBUG_ENTER("count_relay_log_space");
-  rli->log_space_total = 0;
+  rli->log_space_total= 0;
   if (rli->relay_log.find_log_pos(&linfo, NullS, 1))
   {
     sql_print_error("Could not find first log while counting relay log space");
@@ -1631,10 +1625,10 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
     protocol->store((uint32) mi->connect_retry);
     protocol->store(mi->master_log_name, &my_charset_bin);
     protocol->store((ulonglong) mi->master_log_pos);
-    protocol->store(mi->rli.relay_log_name +
-		   dirname_length(mi->rli.relay_log_name), &my_charset_bin);
-    protocol->store((ulonglong) mi->rli.relay_log_pos);
-    protocol->store(mi->rli.master_log_name, &my_charset_bin);
+    protocol->store(mi->rli.group_relay_log_name +
+		   dirname_length(mi->rli.group_relay_log_name), &my_charset_bin);
+    protocol->store((ulonglong) mi->rli.group_relay_log_pos);
+    protocol->store(mi->rli.group_master_log_name, &my_charset_bin);
     protocol->store(mi->slave_running ? "Yes":"No", &my_charset_bin);
     protocol->store(mi->rli.slave_running ? "Yes":"No", &my_charset_bin);
     protocol->store(&replicate_do_db);
@@ -1642,7 +1636,7 @@ int show_master_info(THD* thd, MASTER_INFO* mi)
     protocol->store((uint32) mi->rli.last_slave_errno);
     protocol->store(mi->rli.last_slave_error, &my_charset_bin);
     protocol->store((uint32) mi->rli.slave_skip_counter);
-    protocol->store((ulonglong) mi->rli.master_log_pos);
+    protocol->store((ulonglong) mi->rli.group_master_log_pos);
     protocol->store((ulonglong) mi->rli.log_space_total);
     pthread_mutex_unlock(&mi->rli.data_lock);
     pthread_mutex_unlock(&mi->data_lock);
@@ -1673,17 +1667,15 @@ bool flush_master_info(MASTER_INFO* mi)
 
 
 st_relay_log_info::st_relay_log_info()
-  :info_fd(-1), cur_log_fd(-1), master_log_pos(0), save_temporary_tables(0),
-   cur_log_old_open_count(0), log_space_total(0), ignore_log_space_limit(0),
-   slave_skip_counter(0), abort_pos_wait(0), slave_run_id(0),
-   sql_thd(0), last_slave_errno(0), inited(0), abort_slave(0),
-   slave_running(0), skip_log_purge(0),
-   inside_transaction(0) /* the default is autocommit=1 */
-{
-  relay_log_name[0] = master_log_name[0] = 0;
+  :info_fd(-1), cur_log_fd(-1), save_temporary_tables(0),
+   cur_log_old_open_count(0), group_master_log_pos(0), log_space_total(0),
+   ignore_log_space_limit(0), slave_skip_counter(0), abort_pos_wait(0),
+   slave_run_id(0), sql_thd(0), last_slave_errno(0), inited(0), abort_slave(0),
+   slave_running(0)
+{
+  group_relay_log_name[0]= event_relay_log_name[0]= group_master_log_name[0]= 0;
   last_slave_error[0]=0;
   
-
   bzero(&info_file,sizeof(info_file));
   bzero(&cache_buf, sizeof(cache_buf));
   pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
@@ -1745,8 +1737,8 @@ int st_relay_log_info::wait_for_pos(THD* thd, String* log_name,
   set_timespec(abstime,timeout);
 
   DBUG_ENTER("wait_for_pos");
-  DBUG_PRINT("enter",("master_log_name: '%s'  pos: %lu timeout: %ld",
-                      master_log_name, (ulong) master_log_pos, 
+  DBUG_PRINT("enter",("group_master_log_name: '%s'  pos: %lu timeout: %ld",
+                      group_master_log_name, (ulong) group_master_log_pos, 
                       (long) timeout));
 
   pthread_mutex_lock(&data_lock);
@@ -1796,10 +1788,10 @@ int st_relay_log_info::wait_for_pos(THD* thd, String* log_name,
   {
     bool pos_reached;
     int cmp_result= 0;
-    DBUG_ASSERT(*master_log_name || master_log_pos == 0);
-    if (*master_log_name)
+    DBUG_ASSERT(*group_master_log_name || group_master_log_pos == 0);
+    if (*group_master_log_name)
     {
-      char *basename= master_log_name + dirname_length(master_log_name);
+      char *basename= group_master_log_name + dirname_length(group_master_log_name);
       /*
         First compare the parts before the extension.
         Find the dot in the master's log basename,
@@ -1814,13 +1806,13 @@ int st_relay_log_info::wait_for_pos(THD* thd, String* log_name,
       }
       // Now compare extensions.
       char *q_end;
-      ulong master_log_name_extension= strtoul(q, &q_end, 10);
-      if (master_log_name_extension < log_name_extension)
+      ulong group_master_log_name_extension= strtoul(q, &q_end, 10);
+      if (group_master_log_name_extension < log_name_extension)
         cmp_result = -1 ;
       else
-        cmp_result= (master_log_name_extension > log_name_extension) ? 1 : 0 ;
+        cmp_result= (group_master_log_name_extension > log_name_extension) ? 1 : 0 ;
     }
-    pos_reached = ((!cmp_result && master_log_pos >= (ulonglong)log_pos) ||
+    pos_reached = ((!cmp_result && group_master_log_pos >= (ulonglong)log_pos) ||
                    cmp_result > 0);
     if (pos_reached || thd->killed)
       break;
@@ -2127,7 +2119,7 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli)
 	(rli->slave_skip_counter && type_code != ROTATE_EVENT))
     {
       /* TODO: I/O thread should not even log events with the same server id */
-      rli->inc_pos(ev->get_event_len(),
+      rli->inc_group_relay_log_pos(ev->get_event_len(),
 		   type_code != STOP_EVENT ? ev->log_pos : LL(0),
 		   1/* skip lock*/);
       flush_relay_log_info(rli);
@@ -2497,15 +2489,13 @@ slave_begin:
   rli->abort_slave = 0;
   pthread_mutex_unlock(&rli->run_lock);
   pthread_cond_broadcast(&rli->start_cond);
-  // This should always be set to 0 when the slave thread is started
-  rli->pending = 0;
 
   //tell the I/O thread to take relay_log_space_limit into account from now on
   rli->ignore_log_space_limit= 0;
 
   if (init_relay_log_pos(rli,
-			 rli->relay_log_name,
-			 rli->relay_log_pos,
+			 rli->group_relay_log_name,
+			 rli->group_relay_log_pos,
 			 1 /*need data lock*/, &errmsg))
   {
     sql_print_error("Error initializing relay log position: %s",
@@ -2513,18 +2503,18 @@ slave_begin:
     goto err;
   }
   THD_CHECK_SENTRY(thd);
-  DBUG_ASSERT(rli->relay_log_pos >= BIN_LOG_HEADER_SIZE);
-  DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->relay_log_pos);
+  DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
+  DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
   DBUG_ASSERT(rli->sql_thd == thd);
 
   DBUG_PRINT("master_info",("log_file_name: %s  position: %s",
-			    rli->master_log_name,
-			    llstr(rli->master_log_pos,llbuff)));
+			    rli->group_master_log_name,
+			    llstr(rli->group_master_log_pos,llbuff)));
   if (global_system_variables.log_warnings)
     sql_print_error("Slave SQL thread initialized, starting replication in \
 log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
-		    llstr(rli->master_log_pos,llbuff),rli->relay_log_name,
-		    llstr(rli->relay_log_pos,llbuff1));
+		    llstr(rli->group_master_log_pos,llbuff),rli->group_relay_log_name,
+		    llstr(rli->group_relay_log_pos,llbuff1));
 
   /* Read queries from the IO/THREAD until this thread is killed */
 
@@ -2541,7 +2531,7 @@ log '%s' at position %s, relay log '%s' position: %s", RPL_LOG_NAME,
 Error running query, slave SQL thread aborted. Fix the problem, and restart \
 the slave SQL thread with \"SLAVE START\". We stopped at log \
 '%s' position %s",
-		      RPL_LOG_NAME, llstr(rli->master_log_pos, llbuff));
+		      RPL_LOG_NAME, llstr(rli->group_master_log_pos, llbuff));
       goto err;
     }
   }
@@ -2549,7 +2539,7 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \
   /* Thread stopped. Print the current replication position to the log */
   sql_print_error("Slave SQL thread exiting, replication stopped in log \
  '%s' at position %s",
-		  RPL_LOG_NAME, llstr(rli->master_log_pos,llbuff));
+		  RPL_LOG_NAME, llstr(rli->group_master_log_pos,llbuff));
 
  err:
   VOID(pthread_mutex_lock(&LOCK_thread_count));
@@ -2699,7 +2689,7 @@ err:
     rev			The rotate log event read from the binary log
 
   DESCRIPTION
-    Updates the master info and relay data with the place in the next binary
+    Updates the master info with the place in the next binary
     log where we should start reading.
 
   NOTES
@@ -3073,18 +3063,14 @@ bool flush_relay_log_info(RELAY_LOG_INFO* rli)
   IO_CACHE *file = &rli->info_file;
   char buff[FN_REFLEN*2+22*2+4], *pos;
 
-  /* sql_thd is not set when calling from init_slave() */
-  if ((rli->sql_thd && rli->sql_thd->options & OPTION_BEGIN))
-    return 0;					// Wait for COMMIT
-
   my_b_seek(file, 0L);
-  pos=strmov(buff, rli->relay_log_name);
+  pos=strmov(buff, rli->group_relay_log_name);
   *pos++='\n';
-  pos=longlong2str(rli->relay_log_pos, pos, 10);
+  pos=longlong2str(rli->group_relay_log_pos, pos, 10);
   *pos++='\n';
-  pos=strmov(pos, rli->master_log_name);
+  pos=strmov(pos, rli->group_master_log_name);
   *pos++='\n';
-  pos=longlong2str(rli->master_log_pos, pos, 10);
+  pos=longlong2str(rli->group_master_log_pos, pos, 10);
   *pos='\n';
   if (my_b_write(file, (byte*) buff, (ulong) (pos-buff)+1))
     error=1;
@@ -3107,7 +3093,7 @@ static IO_CACHE *reopen_relay_log(RELAY_LOG_INFO *rli, const char **errmsg)
   DBUG_ENTER("reopen_relay_log");
 
   IO_CACHE *cur_log = rli->cur_log=&rli->cache_buf;
-  if ((rli->cur_log_fd=open_binlog(cur_log,rli->relay_log_name,
+  if ((rli->cur_log_fd=open_binlog(cur_log,rli->event_relay_log_name,
 				   errmsg)) <0)
     DBUG_RETURN(0);
   /*
@@ -3115,7 +3101,7 @@ static IO_CACHE *reopen_relay_log(RELAY_LOG_INFO *rli, const char **errmsg)
     relay_log_pos	Current log pos
     pending		Number of bytes already processed from the event
   */
-  my_b_seek(cur_log,rli->relay_log_pos + rli->pending);
+  my_b_seek(cur_log,rli->event_relay_log_pos);
   DBUG_RETURN(cur_log);
 }
 
@@ -3124,7 +3110,7 @@ Log_event* next_event(RELAY_LOG_INFO* rli)
 {
   Log_event* ev;
   IO_CACHE* cur_log = rli->cur_log;
-  pthread_mutex_t *log_lock = rli->relay_log.get_log_lock();
+  pthread_mutex_t *log_lock = rli->relay_log.get_log_lock(); 
   const char* errmsg=0;
   THD* thd = rli->sql_thd;
   DBUG_ENTER("next_event");
@@ -3173,7 +3159,7 @@ Log_event* next_event(RELAY_LOG_INFO* rli)
       }
     }
     DBUG_ASSERT(my_b_tell(cur_log) >= BIN_LOG_HEADER_SIZE);
-    DBUG_ASSERT(my_b_tell(cur_log) == rli->relay_log_pos + rli->pending);
+    DBUG_ASSERT(my_b_tell(cur_log) == rli->event_relay_log_pos);
     /*
       Relay log is always in new format - if the master is 3.23, the
       I/O thread will convert the format for us
@@ -3240,8 +3226,8 @@ Log_event* next_event(RELAY_LOG_INFO* rli)
         // prevent the I/O thread from blocking next times
         rli->ignore_log_space_limit= 1; 
         // If the I/O thread is blocked, unblock it
-        pthread_cond_broadcast(&rli->log_space_cond);
         pthread_mutex_unlock(&rli->log_space_lock);
+        pthread_cond_broadcast(&rli->log_space_cond);
         // Note that wait_for_update unlocks lock_log !
         rli->relay_log.wait_for_update(rli->sql_thd);
         // re-acquire data lock since we released it earlier
@@ -3258,16 +3244,25 @@ Log_event* next_event(RELAY_LOG_INFO* rli)
       my_close(rli->cur_log_fd, MYF(MY_WME));
       rli->cur_log_fd = -1;
 	
-      /*
-	TODO: make skip_log_purge a start-up option. At this point this
-	is not critical priority
-      */
-      if (!rli->skip_log_purge)
+      if (relay_log_purge)
       {
-	// purge_first_log will properly set up relay log coordinates in rli
-	if (rli->relay_log.purge_first_log(rli))
+	/*
+          purge_first_log will properly set up relay log coordinates in rli.
+          If the group's coordinates are equal to the event's coordinates
+          (i.e. the relay log was not rotated in the middle of a group),
+          we can purge this relay log too.
+          We do ulonglong and string comparisons, this may be slow but
+          - purging the last relay log is nice (it can save 1GB of disk), so we
+          like to detect the case where we can do it, and given this,
+          - I see no better detection method
+          - purge_first_log is not called that often
+        */
+	if (rli->relay_log.purge_first_log
+            (rli,
+             rli->group_relay_log_pos == rli->event_relay_log_pos
+             && !strcmp(rli->group_relay_log_name,rli->event_relay_log_name)))
 	{
-	  errmsg = "Error purging processed log";
+	  errmsg = "Error purging processed logs";
 	  goto err;
 	}
       }
@@ -3285,10 +3280,9 @@ Log_event* next_event(RELAY_LOG_INFO* rli)
 	  errmsg = "error switching to the next log";
 	  goto err;
 	}
-	rli->relay_log_pos = BIN_LOG_HEADER_SIZE;
-	rli->pending=0;
-	strmake(rli->relay_log_name,rli->linfo.log_file_name,
-		sizeof(rli->relay_log_name)-1);
+	rli->event_relay_log_pos = BIN_LOG_HEADER_SIZE;
+	strmake(rli->event_relay_log_name,rli->linfo.log_file_name,
+		sizeof(rli->event_relay_log_name)-1);
 	flush_relay_log_info(rli);
       }
 	
@@ -3336,7 +3330,7 @@ Log_event* next_event(RELAY_LOG_INFO* rli)
 event(errno: %d  cur_log->error: %d)",
 		      my_errno,cur_log->error);
       // set read position to the beginning of the event
-      my_b_seek(cur_log,rli->relay_log_pos+rli->pending);
+      my_b_seek(cur_log,rli->event_relay_log_pos);
       /* otherwise, we have had a partial read */
       errmsg = "Aborting slave SQL thread because of partial event read";
       break;					// To end of function
diff --git a/sql/slave.h b/sql/slave.h
index a4db7388be5..16ba7f80471 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -92,12 +92,6 @@ typedef struct st_relay_log_info
     cur_log_fd - file descriptor of the current read  relay log
   */
   File info_fd,cur_log_fd;
-  /* name of current read relay log */
-  char relay_log_name[FN_REFLEN];
-  /* master log name corresponding to current read position */
-  char master_log_name[FN_REFLEN];
-  /* original log position of last processed event */
-  volatile my_off_t master_log_pos;
 
   /*
     Protected with internal locks.
@@ -142,20 +136,36 @@ typedef struct st_relay_log_info
   uint32 cur_log_old_open_count;
   
   /*
-    relay_log_pos - Current offset in the relay log.
-    pending       - In some cases we do not increment offset immediately
-                    after processing an event, because the following event
-                    needs to be processed atomically together with this one
-                    such as:
-
-                    Intvar_event - sets auto_increment value
-                    Rand_event   - sets the random seed
-
-                    However, once both events have been processed, we need to
-                    increment by the cumulative offset.  'pending' stores the
-                    extra offset to be added to the position.
+    Let's call a group (of events) :
+      - a transaction
+      or
+      - an autocommiting query + its associated events (INSERT_ID,
+    TIMESTAMP...)
+    We need these rli coordinates :
+    - relay log name and position of the beginning of the group we currently are
+    executing. Needed to know where we have to restart when replication has
+    stopped in the middle of a group (which has been rolled back by the slave).
+    - relay log name and position just after the event we have just
+    executed. This event is part of the current group.
+    Formerly we only had the immediately above coordinates, plus a 'pending'
+    variable, but this dealt wrong with the case of a transaction starting on a
+    relay log and finishing (commiting) on another relay log. Case which can
+    happen when, for example, the relay log gets rotated because of
+    max_binlog_size.
+  */
+  char group_relay_log_name[FN_REFLEN];
+  ulonglong group_relay_log_pos;
+  char event_relay_log_name[FN_REFLEN];
+  ulonglong event_relay_log_pos;
+  /* 
+     Original log name and position of the group we're currently executing
+     (whose coordinates are group_relay_log_name/pos in the relay log)
+     in the master's binlog. These concern the *group*, because in the master's
+     binlog the log_pos that comes with each event is the position of the
+     beginning of the group.
   */
-  ulonglong relay_log_pos, pending;
+  char group_master_log_name[FN_REFLEN];
+  volatile my_off_t group_master_log_pos;
 
   /*
     Handling of the relay_log_space_limit optional constraint.
@@ -193,38 +203,39 @@ typedef struct st_relay_log_info
   /* if not set, the value of other members of the structure are undefined */
   bool inited;
   volatile bool abort_slave, slave_running;
-  bool skip_log_purge;
-  bool inside_transaction;
 
   st_relay_log_info();
   ~st_relay_log_info();
-  inline void inc_pending(ulonglong val)
+
+  inline void inc_event_relay_log_pos(ulonglong val)
   {
-    pending += val;
+    event_relay_log_pos+= val;
   }
-  /* TODO: this probably needs to be fixed */
-  inline void inc_pos(ulonglong val, ulonglong log_pos, bool skip_lock=0)
+
+  void inc_group_relay_log_pos(ulonglong val, ulonglong log_pos, bool skip_lock=0)
   {
     if (!skip_lock)
       pthread_mutex_lock(&data_lock);
-    relay_log_pos += val+pending;
-    pending = 0;
-    if (log_pos)
-      master_log_pos = log_pos+ val;
+    inc_event_relay_log_pos(val);
+    group_relay_log_pos= event_relay_log_pos;
+    strmake(group_relay_log_name,event_relay_log_name,
+            sizeof(group_relay_log_name)-1);
+    /*
+      If the slave does not support transactions and replicates a transaction,
+      users should not trust group_master_log_pos (which they can display with
+      SHOW SLAVE STATUS or read from relay-log.info), because to compute
+      group_master_log_pos the slave relies on log_pos stored in the master's
+      binlog, but if we are in a master's transaction these positions are always
+      the BEGIN's one (excepted for the COMMIT), so group_master_log_pos does
+      not advance as it should on the non-transactional slave (it advances by
+      big leaps, whereas it should advance by small leaps).
+    */
+    if (log_pos) // 3.23 binlogs don't have log_posx
+      group_master_log_pos= log_pos+ val;
     pthread_cond_broadcast(&data_cond);
     if (!skip_lock)
       pthread_mutex_unlock(&data_lock);
   }
-  /*
-    thread safe read of position - not needed if we are in the slave thread,
-    but required otherwise as var is a longlong
-  */
-  inline void read_pos(ulonglong& var)
-  {
-    pthread_mutex_lock(&data_lock);
-    var = relay_log_pos;
-    pthread_mutex_unlock(&data_lock);
-  }
 
   int wait_for_pos(THD* thd, String* log_name, longlong log_pos, 
 		   longlong timeout);
@@ -334,7 +345,7 @@ typedef struct st_table_rule_ent
 #define TABLE_RULE_ARR_SIZE   16
 #define MAX_SLAVE_ERRMSG      1024
 
-#define RPL_LOG_NAME (rli->master_log_name[0] ? rli->master_log_name :\
+#define RPL_LOG_NAME (rli->group_master_log_name[0] ? rli->group_master_log_name :\
  "FIRST")
 #define IO_RPL_LOG_NAME (mi->master_log_name[0] ? mi->master_log_name :\
  "FIRST")
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 2063e8b3f08..e374cdb3696 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -49,7 +49,7 @@ extern "C" byte *table_cache_key(const byte *record,uint *length,
 
 void table_cache_init(void)
 {
-  VOID(hash_init(&open_cache,system_charset_info,
+  VOID(hash_init(&open_cache,&my_charset_bin,
 		 table_cache_size+16,0,0,table_cache_key,
 		 (hash_free_key) free_cache_entry,0));
   mysql_rm_tmp_tables();
@@ -245,16 +245,11 @@ static void free_cache_entry(TABLE *table)
 void free_io_cache(TABLE *table)
 {
   DBUG_ENTER("free_io_cache");
-  if (table->io_cache)
+  if (table->sort.io_cache)
   {
-    close_cached_file(table->io_cache);
-    my_free((gptr) table->io_cache,MYF(0));
-    table->io_cache=0;
-  }
-  if (table->record_pointers)
-  {
-    my_free((gptr) table->record_pointers,MYF(0));
-    table->record_pointers=0;
+    close_cached_file(table->sort.io_cache);
+    my_free((gptr) table->sort.io_cache,MYF(0));
+    table->sort.io_cache=0;
   }
   DBUG_VOID_RETURN;
 }
diff --git a/sql/sql_cache.cc b/sql/sql_cache.cc
index 912cce3ec91..d8265a1b359 100644
--- a/sql/sql_cache.cc
+++ b/sql/sql_cache.cc
@@ -758,9 +758,11 @@ void Query_cache::store_query(THD *thd, TABLE_LIST *tables_used)
   DBUG_ENTER("Query_cache::store_query");
   if (query_cache_size == 0)
     DBUG_VOID_RETURN;
+  uint8 tables_type= 0;
 
   if ((local_tables = is_cacheable(thd, thd->query_length,
-			     thd->query, &thd->lex, tables_used)))
+			     thd->query, &thd->lex, tables_used,
+			     &tables_type)))
   {
     NET *net= &thd->net;
     byte flags = (thd->client_capabilities & CLIENT_LONG_FLAG ? 0x80 : 0);
@@ -836,6 +838,7 @@ void Query_cache::store_query(THD *thd, TABLE_LIST *tables_used)
 
 	net->query_cache_query= (gptr) query_block;
 	header->writer(net);
+	header->tables_type(tables_type);
 	// init_n_lock make query block locked
 	BLOCK_UNLOCK_WR(query_block);
       }
@@ -882,15 +885,10 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
   Query_cache_block_table *block_table, *block_table_end;
   ulong tot_length;
   byte flags;
+  bool check_tables;
   DBUG_ENTER("Query_cache::send_result_to_client");
 
-  if (query_cache_size == 0 ||
-      /*
-	it is not possible to check has_transactions() function of handler
-	because tables not opened yet
-      */
-      (thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) ||
-      thd->variables.query_cache_type == 0)
+  if (query_cache_size == 0 || thd->variables.query_cache_type == 0)
 
     goto err;
 
@@ -970,6 +968,16 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
   }
   DBUG_PRINT("qcache", ("Query have result 0x%lx", (ulong) query));
 
+  if ((thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) &&
+      (query->tables_type() & HA_CACHE_TBL_TRANSACT))
+  {
+    DBUG_PRINT("qcache",
+	       ("we are in transaction and have transaction tables in query"));
+    BLOCK_UNLOCK_RD(query_block);
+    goto err_unlock;
+  }
+      
+  check_tables= query->tables_type() & HA_CACHE_TBL_ASKTRANSACT;
   // Check access;
   block_table= query_block->table(0);
   block_table_end= block_table+query_block->n_tables;
@@ -997,9 +1005,22 @@ Query_cache::send_result_to_client(THD *thd, char *sql, uint query_length)
       DBUG_PRINT("qcache", ("Need to check column privileges for %s.%s",
 			    table_list.db, table_list.alias));
       BLOCK_UNLOCK_RD(query_block);
-      thd->lex.safe_to_cache_query=0;		// Don't try to cache this
+      thd->lex.safe_to_cache_query= 0;		// Don't try to cache this
       goto err_unlock;				// Parse query
     }
+    if (check_tables && !handler::caching_allowed(thd, table->db(), 
+						  table->key_length(),
+						  table->type()))
+    {
+      DBUG_PRINT("qcache", ("Handler does not allow caching for %s.%s",
+			    table_list.db, table_list.alias));
+      BLOCK_UNLOCK_RD(query_block);
+      thd->lex.safe_to_cache_query= 0;          // Don't try to cache this
+      goto err_unlock;				// Parse query
+    }
+    else
+      DBUG_PRINT("qcache", ("handler allow caching (%d) %s,%s",
+			    check_tables, table_list.db, table_list.alias));
   }
   move_to_query_list_end(query_block);
   hits++;
@@ -1061,7 +1082,8 @@ void Query_cache::invalidate(THD *thd, TABLE_LIST *tables_used,
 	if (tables_used->derived)
 	  continue;
 	if (using_transactions && 
-	   tables_used->table->file->has_transactions())
+	   (tables_used->table->file->table_cache_type() == 
+	    HA_CACHE_TBL_TRANSACT))
 	  /* 
 	     Tables_used->table can't be 0 in transaction.
 	     Only 'drop' invalidate not opened table, but 'drop' 
@@ -1115,7 +1137,8 @@ void Query_cache::invalidate(THD *thd, TABLE *table,
     {
       using_transactions = using_transactions &&
 	(thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN));
-      if (using_transactions && table->file->has_transactions())
+      if (using_transactions && 
+	  (table->file->table_cache_type() == HA_CACHE_TBL_TRANSACT))
 	thd->add_changed_table(table);
       else
 	invalidate_table(table);
@@ -1934,7 +1957,8 @@ my_bool Query_cache::register_all_tables(Query_cache_block *block,
     block_table->n=n;
     if (!insert_table(tables_used->table->key_length,
 		      tables_used->table->table_cache_key, block_table,
-		      tables_used->db_length))
+		      tables_used->db_length,
+		      tables_used->table->file->table_cache_type()))
       break;
 
     if (tables_used->table->db_type == DB_TYPE_MRG_MYISAM)
@@ -1947,11 +1971,12 @@ my_bool Query_cache::register_all_tables(Query_cache_block *block,
       {
 	char key[MAX_DBKEY_LENGTH];
 	uint32 db_length;
-	uint key_length =filename_2_table_key(key, table->table->filename,
+	uint key_length= filename_2_table_key(key, table->table->filename,
 					      &db_length);
 	(++block_table)->n= ++n;
 	if (!insert_table(key_length, key, block_table,
-			  db_length))
+			  db_length,
+			  tables_used->table->file->table_cache_type()))
 	  goto err;
       }
     }
@@ -1978,7 +2003,7 @@ err:
 my_bool
 Query_cache::insert_table(uint key_len, char *key,
 			  Query_cache_block_table *node,
-			  uint32 db_length)
+			  uint32 db_length, uint8 cache_type)
 {
   DBUG_ENTER("Query_cache::insert_table");
   DBUG_PRINT("qcache", ("insert table node 0x%lx, len %d",
@@ -2016,6 +2041,8 @@ Query_cache::insert_table(uint key_len, char *key,
     }
     char *db = header->db();
     header->table(db + db_length + 1);
+    header->key_length(key_len);
+    header->type(cache_type);
   }
 
   Query_cache_block_table *list_root = table_block->table(0);
@@ -2446,7 +2473,9 @@ void Query_cache::double_linked_list_join(Query_cache_block *head_tail,
 
 TABLE_COUNTER_TYPE Query_cache::is_cacheable(THD *thd, uint32 query_len,
 					     char *query,
-					     LEX *lex, TABLE_LIST *tables_used)
+					     LEX *lex,
+					     TABLE_LIST *tables_used,
+					     uint8 *tables_type)
 {
   TABLE_COUNTER_TYPE table_count = 0;
   DBUG_ENTER("Query_cache::is_cacheable");
@@ -2457,7 +2486,6 @@ TABLE_COUNTER_TYPE Query_cache::is_cacheable(THD *thd, uint32 query_len,
 						 OPTION_TO_QUERY_CACHE))) &&
       lex->safe_to_cache_query)
   {
-    my_bool has_transactions = 0;
     DBUG_PRINT("qcache", ("options %lx %lx, type %u",
 			OPTION_TO_QUERY_CACHE,
 			lex->select_lex.options,
@@ -2469,8 +2497,7 @@ TABLE_COUNTER_TYPE Query_cache::is_cacheable(THD *thd, uint32 query_len,
       DBUG_PRINT("qcache", ("table %s, db %s, type %u",
 			  tables_used->real_name,
 			  tables_used->db, tables_used->table->db_type));
-      has_transactions = (has_transactions ||
-			  tables_used->table->file->has_transactions());
+      *tables_type|= tables_used->table->file->table_cache_type();
 
       if (tables_used->table->db_type == DB_TYPE_MRG_ISAM ||
 	  tables_used->table->tmp_table != NO_TMP_TABLE ||
@@ -2500,7 +2527,7 @@ TABLE_COUNTER_TYPE Query_cache::is_cacheable(THD *thd, uint32 query_len,
     }
 
     if ((thd->options & (OPTION_NOT_AUTOCOMMIT | OPTION_BEGIN)) &&
-	has_transactions)
+	((*tables_type)&HA_CACHE_TBL_TRANSACT))
     {
       DBUG_PRINT("qcache", ("not in autocommin mode"));
       DBUG_RETURN(0);
@@ -2941,7 +2968,7 @@ void Query_cache::bins_dump()
 {
   uint i;
   
-  if (!initialized)
+  if (!initialized || query_cache_size == 0)
   {
     DBUG_PRINT("qcache", ("Query Cache not initialized"));
     return;
@@ -2982,7 +3009,7 @@ void Query_cache::bins_dump()
 
 void Query_cache::cache_dump()
 {
-  if (!initialized)
+  if (!initialized || query_cache_size == 0)
   {
     DBUG_PRINT("qcache", ("Query Cache not initialized"));
     return;
@@ -3071,7 +3098,7 @@ void Query_cache::queries_dump()
 
 void Query_cache::tables_dump()
 {
-  if (!initialized)
+  if (!initialized || query_cache_size == 0)
   {
     DBUG_PRINT("qcache", ("Query Cache not initialized"));
     return;
diff --git a/sql/sql_cache.h b/sql/sql_cache.h
index 68bd3d48ff5..eea542e9d06 100644
--- a/sql/sql_cache.h
+++ b/sql/sql_cache.h
@@ -115,18 +115,21 @@ struct Query_cache_query
   Query_cache_block *res;
   NET *wri;
   ulong len;
+  uint8 tbls_type;
 
   inline void init_n_lock();
   void unlock_n_destroy();
   inline ulonglong found_rows()		   { return limit_found_rows; }
-  inline void found_rows(ulonglong rows)   { limit_found_rows = rows; }
+  inline void found_rows(ulonglong rows)   { limit_found_rows= rows; }
   inline Query_cache_block *result()	   { return res; }
-  inline void result(Query_cache_block *p) { res=p; }
+  inline void result(Query_cache_block *p) { res= p; }
   inline NET *writer()			   { return wri; }
-  inline void writer(NET *p)		   { wri=p; }
+  inline void writer(NET *p)		   { wri= p; }
+  inline uint8 tables_type()               { return tbls_type; }
+  inline void tables_type(uint8 type)      { tbls_type= type; }
   inline ulong length()			   { return len; }
-  inline ulong add(ulong packet_len)	   { return(len += packet_len); }
-  inline void length(ulong length)	   { len = length; }
+  inline ulong add(ulong packet_len)	   { return(len+= packet_len); }
+  inline void length(ulong length)	   { len= length; }
   inline gptr query()
   {
     return (gptr)(((byte*)this)+
@@ -144,10 +147,16 @@ struct Query_cache_query
 struct Query_cache_table
 {
   char *tbl;
+  uint key_len;
+  uint8 table_type;
 
   inline char *db()			     { return (char *) data(); }
   inline char *table()			     { return tbl; }
-  inline void table(char *table)	     { tbl = table; }
+  inline void table(char *table)	     { tbl= table; }
+  inline uint key_length()                   { return key_len; }
+  inline void key_length(uint len)           { key_len= len; }
+  inline uint8 type()                        { return table_type; }
+  inline void type(uint8 t)                  { table_type= t; }
   inline gptr data()
   {
     return (gptr)(((byte*)this)+
@@ -276,7 +285,7 @@ protected:
 			      TABLE_COUNTER_TYPE tables);
   my_bool insert_table(uint key_len, char *key,
 		       Query_cache_block_table *node,
-		       uint32 db_length);
+		       uint32 db_length, uint8 cache_type);
   void unlink_table(Query_cache_block_table *node);
   Query_cache_block *get_free_block (ulong len, my_bool not_less,
 				      ulong min);
@@ -334,7 +343,8 @@ protected:
     (query without tables not cached)
   */
   TABLE_COUNTER_TYPE is_cacheable(THD *thd, uint32 query_len, char *query,
-				  LEX *lex, TABLE_LIST *tables_used);
+				  LEX *lex, TABLE_LIST *tables_used,
+				  uint8 *tables_type);
  public:
 
   Query_cache(ulong query_cache_limit = ULONG_MAX,
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 31159dc259f..f579b02ee50 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -207,7 +207,6 @@ void THD::init(void)
 {
   pthread_mutex_lock(&LOCK_global_system_variables);
   variables= global_system_variables;
-  variables.client_collation= default_charset_info;
   pthread_mutex_unlock(&LOCK_global_system_variables);
   server_status= SERVER_STATUS_AUTOCOMMIT;
   options= thd_startup_options;
diff --git a/sql/sql_class.h b/sql/sql_class.h
index ce109035b2a..925afde2202 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -145,10 +145,12 @@ public:
   int generate_new_name(char *new_name,const char *old_name);
   void make_log_name(char* buf, const char* log_ident);
   bool is_active(const char* log_file_name);
-  int update_log_index(LOG_INFO* linfo);
-  int purge_logs(THD* thd, const char* to_log);
-  int purge_logs_before_date(THD* thd, time_t purge_time);
-  int purge_first_log(struct st_relay_log_info* rli); 
+  int update_log_index(LOG_INFO* linfo, bool need_update_threads);
+  int purge_logs(const char *to_log, bool included, 
+                 bool need_mutex, bool need_update_threads,
+                 ulonglong *decrease_log_space);
+  int purge_logs_before_date(time_t purge_time);
+  int purge_first_log(struct st_relay_log_info* rli, bool included); 
   bool reset_logs(THD* thd);
   // if we are exiting, we also want to close the index file
   void close(bool exiting = 0);
@@ -352,6 +354,7 @@ struct system_variables
   ulong max_allowed_packet;
   ulong max_error_count;
   ulong max_heap_table_size;
+  ulong max_length_for_sort_data;
   ulong max_prep_stmt_count;
   ulong max_sort_length;
   ulong max_tmp_tables;
@@ -382,9 +385,9 @@ struct system_variables
   my_bool low_priority_updates;
   my_bool new_mode;
   
-  CHARSET_INFO 	*client_collation;
-  CHARSET_INFO  *literal_collation;
-  CHARSET_INFO  *result_collation;
+  CHARSET_INFO 	*collation_client;
+  CHARSET_INFO  *collation_connection;
+  CHARSET_INFO  *collation_results;
 };
 
 void free_tmp_table(THD *thd, TABLE *entry);
@@ -665,7 +668,7 @@ public:
     net.report_error= 1; 
     DBUG_PRINT("error",("Fatal error set"));
   }
-  inline CHARSET_INFO *charset() { return variables.client_collation; }
+  inline CHARSET_INFO *charset() { return variables.collation_client; }
 };
 
 /*
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index ec77668c968..9bc4dfc74e7 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1203,23 +1203,60 @@ TABLE_LIST *st_select_lex_node::add_table_to_list(THD *thd, Table_ident *table,
 }
 ulong st_select_lex_node::get_table_join_options() { return 0; }
 
-/*
-  This is used for UNION & subselect to create a new table list of all used 
-  tables.
-  The table_list->table entry in all used tables are set to point
-  to the entries in this list.
-*/
 
-// interface
+/*  
+  Interface method of table list creation for query
+  
+  SYNOPSIS
+    st_select_lex_unit::create_total_list()
+    thd            THD pointer
+    result         pointer on result list of tables pointer
+    check_derived  force derived table chacking (used for creating 
+                   table list for derived query)
+  DESCRIPTION
+    This is used for UNION & subselect to create a new table list of all used 
+    tables.
+    The table_list->table entry in all used tables are set to point
+    to the entries in this list.
+
+  RETURN
+    0 - OK
+    !0 - error
+*/
 bool st_select_lex_unit::create_total_list(THD *thd, st_lex *lex,
 					   TABLE_LIST **result,
 					   bool check_derived)
 {
   *result= 0;
-  return create_total_list_n_last_return(thd, lex, &result, check_derived);
+  for (SELECT_LEX_UNIT *unit= this; unit; unit= unit->next_unit())
+  {
+    if ((res= unit->create_total_list_n_last_return(thd, lex, &result,
+						    check_derived)))
+      return res;
+  }
+  return 0;
 }
 
-// list creator
+/*  
+  Table list creation for query
+  
+  SYNOPSIS
+    st_select_lex_unit::create_total_list()
+    thd            THD pointer
+    lex            pointer on LEX stricture
+    result         pointer on pointer on result list of tables pointer
+    check_derived  force derived table chacking (used for creating 
+                   table list for derived query)
+  DESCRIPTION
+    This is used for UNION & subselect to create a new table list of all used 
+    tables.
+    The table_list->table entry in all used tables are set to point
+    to the entries in this list.
+
+  RETURN
+    0 - OK
+    !0 - error
+*/
 bool st_select_lex_unit::create_total_list_n_last_return(THD *thd, st_lex *lex,
 							 TABLE_LIST ***result,
 							 bool check_derived)
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 47743685890..e03814bcd2f 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -230,6 +230,7 @@ public:
 
   virtual st_select_lex_unit* master_unit()= 0;
   virtual st_select_lex* outer_select()= 0;
+  virtual st_select_lex_node* return_after_parsing()= 0;
 
   virtual bool set_braces(bool value);
   virtual bool inc_in_sum_expr();
@@ -284,6 +285,8 @@ public:
     global parameters for union
   */
   st_select_lex_node *global_parameters;
+  //node on wich we should return current_select pointer after parsing subquery
+  st_select_lex_node *return_to;
   /* LIMIT clause runtime counters */
   ha_rows select_limit_cnt, offset_limit_cnt;
   /* not NULL if union used in subselect, point to subselect item */
@@ -304,6 +307,7 @@ public:
       (st_select_lex*) slave->next : (st_select_lex*) slave;
   }
   st_select_lex_unit* next_unit() { return (st_select_lex_unit*) next; }
+  st_select_lex_node* return_after_parsing() { return return_to; }
   void exclude_level();
 
   /* UNION methods */
@@ -366,6 +370,10 @@ public:
   {
     return &link_next;
   }
+  st_select_lex_node* return_after_parsing()
+  {
+    return master_unit()->return_after_parsing();
+  }
 
   bool set_braces(bool value);
   bool inc_in_sum_expr();
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 2f4915c74f1..ba8a4af794a 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -3235,7 +3235,8 @@ mysql_init_query(THD *thd)
   lex->select_lex.init_query();
   lex->value_list.empty();
   lex->param_list.empty();
-  lex->unit.next= lex->unit.master= lex->unit.link_next= 0;
+  lex->unit.next= lex->unit.master= lex->unit.return_to= 
+    lex->unit.link_next= 0;
   lex->unit.prev= lex->unit.link_prev= 0;
   lex->unit.global_parameters= lex->unit.slave= lex->current_select=
     lex->all_selects_list= &lex->select_lex;
@@ -3283,9 +3284,9 @@ bool
 mysql_new_select(LEX *lex, bool move_down)
 {
   SELECT_LEX *select_lex = new SELECT_LEX();
-  select_lex->select_number= ++lex->thd->select_number;
   if (!select_lex)
     return 1;
+  select_lex->select_number= ++lex->thd->select_number;
   select_lex->init_query();
   select_lex->init_select();
   if (move_down)
@@ -3297,9 +3298,13 @@ mysql_new_select(LEX *lex, bool move_down)
     unit->init_query();
     unit->init_select();
     unit->thd= lex->thd;
-    unit->include_down(lex->current_select);
+    if (lex->current_select->linkage == GLOBAL_OPTIONS_TYPE)
+      unit->include_neighbour(lex->current_select);
+    else
+      unit->include_down(lex->current_select);
     unit->link_next= 0;
     unit->link_prev= 0;
+    unit->return_to= lex->current_select;
     select_lex->include_down(unit);
   }
   else
@@ -3975,7 +3980,7 @@ bool reload_acl_and_cache(THD *thd, ulong options, TABLE_LIST *tables)
     {
       long purge_time= time(0) - expire_logs_days*24*60*60;
       if (purge_time >= 0)
-	mysql_bin_log.purge_logs_before_date(thd, purge_time);
+	mysql_bin_log.purge_logs_before_date(purge_time);
     }
 #endif
     mysql_slow_log.new_file(1);
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 7e9b6aea7b5..0eb444b85c0 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -292,7 +292,7 @@ int purge_master_logs(THD* thd, const char* to_log)
   char search_file_name[FN_REFLEN];
 
   mysql_bin_log.make_log_name(search_file_name, to_log);
-  int res = mysql_bin_log.purge_logs(thd, search_file_name);
+  int res = mysql_bin_log.purge_logs(search_file_name, 0, 1, 1, NULL);
 
   return purge_error_message(thd, res);
 }
@@ -300,7 +300,7 @@ int purge_master_logs(THD* thd, const char* to_log)
 
 int purge_master_logs_before_date(THD* thd, time_t purge_time)
 {
-  int res = mysql_bin_log.purge_logs_before_date(thd, purge_time);
+  int res = mysql_bin_log.purge_logs_before_date(purge_time);
   return purge_error_message(thd ,res);
 }
 
@@ -776,24 +776,25 @@ int reset_slave(THD *thd, MASTER_INFO* mi)
     error=1;
     goto err;
   }
-  //delete relay logs, clear relay log coordinates
+  // delete relay logs, clear relay log coordinates
   if ((error= purge_relay_logs(&mi->rli, thd,
 			       1 /* just reset */,
 			       &errmsg)))
     goto err;
   
-  //Clear master's log coordinates (only for good display of SHOW SLAVE STATUS)
+  // Clear master's log coordinates (only for good display of SHOW SLAVE STATUS)
   mi->master_log_name[0]= 0;
   mi->master_log_pos= BIN_LOG_HEADER_SIZE;
-  //close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
+  // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
   end_master_info(mi);
-  //and delete these two files
+  // and delete these two files
   fn_format(fname, master_info_file, mysql_data_home, "", 4+32);
   if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME)))
   {
     error=1;
     goto err;
   }
+  // delete relay_log_info_file
   fn_format(fname, relay_log_info_file, mysql_data_home, "", 4+32);
   if (my_stat(fname, &stat_area, MYF(0)) && my_delete(fname, MYF(MY_WME)))
   {
@@ -874,7 +875,6 @@ int change_master(THD* thd, MASTER_INFO* mi)
     // if we change host or port, we must reset the postion
     mi->master_log_name[0] = 0;
     mi->master_log_pos= BIN_LOG_HEADER_SIZE;
-    mi->rli.pending = 0;
   }
 
   if (lex_mi->log_file_name)
@@ -883,7 +883,6 @@ int change_master(THD* thd, MASTER_INFO* mi)
   if (lex_mi->pos)
   {
     mi->master_log_pos= lex_mi->pos;
-    mi->rli.pending = 0;
   }
   DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos));
 
@@ -901,20 +900,22 @@ int change_master(THD* thd, MASTER_INFO* mi)
   if (lex_mi->relay_log_name)
   {
     need_relay_log_purge= 0;
-    strmake(mi->rli.relay_log_name,lex_mi->relay_log_name,
-	    sizeof(mi->rli.relay_log_name)-1);
+    strmake(mi->rli.group_relay_log_name,lex_mi->relay_log_name,
+	    sizeof(mi->rli.group_relay_log_name)-1);
+    strmake(mi->rli.event_relay_log_name,lex_mi->relay_log_name,
+	    sizeof(mi->rli.event_relay_log_name)-1);
   }
 
   if (lex_mi->relay_log_pos)
   {
     need_relay_log_purge= 0;
-    mi->rli.relay_log_pos=lex_mi->relay_log_pos;
+    mi->rli.group_relay_log_pos= mi->rli.event_relay_log_pos= lex_mi->relay_log_pos;
   }
 
   flush_master_info(mi);
   if (need_relay_log_purge)
   {
-    mi->rli.skip_log_purge= 0;
+    relay_log_purge= 1;
     thd->proc_info="purging old relay logs";
     if (purge_relay_logs(&mi->rli, thd,
 			 0 /* not only reset, but also reinit */,
@@ -928,11 +929,11 @@ int change_master(THD* thd, MASTER_INFO* mi)
   else
   {
     const char* msg;
-    mi->rli.skip_log_purge= 1;
+    relay_log_purge= 0;
     /* Relay log is already initialized */
     if (init_relay_log_pos(&mi->rli,
-			   mi->rli.relay_log_name,
-			   mi->rli.relay_log_pos,
+			   mi->rli.group_relay_log_name,
+			   mi->rli.group_relay_log_pos,
 			   0 /*no data lock*/,
 			   &msg))
     {
@@ -941,12 +942,12 @@ int change_master(THD* thd, MASTER_INFO* mi)
       DBUG_RETURN(1);
     }
   }
-  mi->rli.master_log_pos = mi->master_log_pos;
+  mi->rli.group_master_log_pos = mi->master_log_pos;
   DBUG_PRINT("info", ("master_log_pos: %d", (ulong) mi->master_log_pos));
-  strmake(mi->rli.master_log_name,mi->master_log_name,
-	  sizeof(mi->rli.master_log_name)-1);
-  if (!mi->rli.master_log_name[0]) // uninitialized case
-    mi->rli.master_log_pos=0;
+  strmake(mi->rli.group_master_log_name,mi->master_log_name,
+	  sizeof(mi->rli.group_master_log_name)-1);
+  if (!mi->rli.group_master_log_name[0]) // uninitialized case
+    mi->rli.group_master_log_pos=0;
 
   pthread_mutex_lock(&mi->rli.data_lock);
   mi->rli.abort_pos_wait++;
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 55567493a4a..18768099f0f 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -299,7 +299,7 @@ JOIN::prepare(Item ***rref_pointer_array,
     DBUG_RETURN(-1);				/* purecov: inspected */
 
   ref_pointer_array= *rref_pointer_array;
-
+  
   if (having)
   {
     thd->where="having clause";
@@ -313,6 +313,7 @@ JOIN::prepare(Item ***rref_pointer_array,
     if (having->with_sum_func)
       having->split_sum_func(ref_pointer_array, all_fields);
   }
+  
   if (setup_ftfuncs(select_lex)) /* should be after having->fix_fields */
     DBUG_RETURN(-1);
   /*
@@ -426,7 +427,7 @@ JOIN::optimize()
 
 #ifdef HAVE_REF_TO_FIELDS			// Not done yet
   /* Add HAVING to WHERE if possible */
-  if (having && !group_list && ! sum_func_count)
+  if (having && !group_list && !sum_func_count)
   {
     if (!conds)
     {
@@ -4014,7 +4015,7 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
     case Item_sum::AVG_FUNC:			/* Place for sum & count */
       if (group)
 	return new Field_string(sizeof(double)+sizeof(longlong),
-				maybe_null, item->name,table,&my_charset_bin);
+				0, item->name,table,&my_charset_bin);
       else
 	return new Field_double(item_sum->max_length,maybe_null,
 				item->name, table, item_sum->decimals);
@@ -4022,7 +4023,7 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
     case Item_sum::STD_FUNC:	
       if (group)
 	return	new Field_string(sizeof(double)*2+sizeof(longlong),
-				 maybe_null, item->name,table,&my_charset_bin);
+				 0, item->name,table,&my_charset_bin);
       else
 	return new Field_double(item_sum->max_length, maybe_null,
 				item->name,table,item_sum->decimals);				
@@ -5626,11 +5627,11 @@ end_send(JOIN *join, JOIN_TAB *join_tab __attribute__((unused)),
 	  TABLE *table=jt->table;
 
 	  join->select_options ^= OPTION_FOUND_ROWS;
-	  if (table->record_pointers ||
-	      (table->io_cache && my_b_inited(table->io_cache)))
+	  if (table->sort.record_pointers ||
+	      (table->sort.io_cache && my_b_inited(table->sort.io_cache)))
 	  {
 	    /* Using filesort */
-	    join->send_records= table->found_records;
+	    join->send_records= table->sort.found_records;
 	  }
 	  else
 	  {
@@ -6465,8 +6466,8 @@ create_sort_index(THD *thd, JOIN_TAB *tab, ORDER *order,
   if (!(sortorder=make_unireg_sortorder(order,&length)))
     goto err;				/* purecov: inspected */
   /* It's not fatal if the following alloc fails */
-  table->io_cache=(IO_CACHE*) my_malloc(sizeof(IO_CACHE),
-					MYF(MY_WME | MY_ZEROFILL));
+  table->sort.io_cache=(IO_CACHE*) my_malloc(sizeof(IO_CACHE),
+                                             MYF(MY_WME | MY_ZEROFILL));
   table->status=0;				// May be wrong if quick_select
 
   // If table has a range, move it to select
@@ -6495,9 +6496,9 @@ create_sort_index(THD *thd, JOIN_TAB *tab, ORDER *order,
   }
   if (table->tmp_table)
     table->file->info(HA_STATUS_VARIABLE);	// Get record count
-  table->found_records=filesort(thd, table,sortorder, length,
-				select, filesort_limit, &examined_rows);
-  tab->records=table->found_records;		// For SQL_CALC_ROWS
+  table->sort.found_records=filesort(thd, table,sortorder, length,
+                                     select, filesort_limit, &examined_rows);
+  tab->records=table->sort.found_records;		// For SQL_CALC_ROWS
   delete select;				// filesort did select
   tab->select=0;
   tab->select_cond=0;
@@ -6509,7 +6510,7 @@ create_sort_index(THD *thd, JOIN_TAB *tab, ORDER *order,
     table->key_read=0;
     table->file->extra(HA_EXTRA_NO_KEYREAD);
   }
-  DBUG_RETURN(table->found_records == HA_POS_ERROR);
+  DBUG_RETURN(table->sort.found_records == HA_POS_ERROR);
 err:
   DBUG_RETURN(-1);
 }
diff --git a/sql/sql_sort.h b/sql/sql_sort.h
index 14463a67a28..9f95ffa4884 100644
--- a/sql/sql_sort.h
+++ b/sql/sql_sort.h
@@ -19,6 +19,30 @@
 #define MERGEBUFF		7
 #define MERGEBUFF2		15
 
+/*
+   The structure SORT_ADDON_FIELD describes a fixed layout
+   for field values appended to sorted values in records to be sorted
+   in the sort buffer.
+   Only fixed layout is supported now.
+   Null bit maps for the appended values is placed before the values 
+   themselves. Offsets are from the last sorted field, that is from the
+   record referefence, which is still last component of sorted records.
+   It is preserved for backward compatiblility.
+   The structure is used tp store values of the additional fields 
+   in the sort buffer. It is used also when these values are read
+   from a temporary file/buffer. As the reading procedures are beyond the
+   scope of the 'filesort' code the values have to be retrieved via
+   the callback function 'unpack_addon_fields'.
+*/
+
+typedef struct st_sort_addon_field {  /* Sort addon packed field */
+  Field *field;          /* Original field */
+  uint   offset;         /* Offset from the last sorted field */
+  uint   null_offset;    /* Offset to to null bit from the last sorted field */
+  uint   length;         /* Length in the sort buffer */
+  uint8  null_bit;       /* Null bit mask for the field */
+} SORT_ADDON_FIELD;
+
 typedef struct st_buffpek {		/* Struktur om sorteringsbuffrarna */
   my_off_t file_pos;			/* Where we are in the sort file */
   uchar *base,*key;			/* key pointers */
@@ -27,15 +51,18 @@ typedef struct st_buffpek {		/* Struktur om sorteringsbuffrarna */
   ulong max_keys;			/* Max keys in buffert */
 } BUFFPEK;
 
-
 typedef struct st_sort_param {
-  uint sort_length;			/* Length of sort columns */
-  uint keys;				/* Max keys / buffert */
+  uint rec_length;          /* Length of sorted records */
+  uint sort_length;			/* Length of sorted columns */
   uint ref_length;			/* Length of record ref. */
+  uint addon_length;        /* Length of added packed fields */
+  uint res_length;          /* Length of records in final sorted file/buffer */
+  uint keys;				/* Max keys / buffer */
   ha_rows max_rows,examined_rows;
   TABLE *sort_form;			/* For quicker make_sortkey */
   SORT_FIELD *local_sortorder;
   SORT_FIELD *end;
+  SORT_ADDON_FIELD *addon_field; /* Descriptors for companion fields */
   uchar *unique_buff;
   bool not_killable;
   char* tmp_buffer;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 36056851753..8d216f121ab 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2348,8 +2348,8 @@ copy_data_between_tables(TABLE *from,TABLE *to,
 
   if (order)
   {
-    from->io_cache=(IO_CACHE*) my_malloc(sizeof(IO_CACHE),
-                                         MYF(MY_FAE | MY_ZEROFILL));
+    from->sort.io_cache=(IO_CACHE*) my_malloc(sizeof(IO_CACHE),
+                                              MYF(MY_FAE | MY_ZEROFILL));
     bzero((char*) &tables,sizeof(tables));
     tables.table = from;
     tables.alias = tables.real_name= from->real_name;
@@ -2361,9 +2361,9 @@ copy_data_between_tables(TABLE *from,TABLE *to,
 	setup_order(thd, thd->lex.select_lex.ref_pointer_array,
 		    &tables, fields, all_fields, order) ||
         !(sortorder=make_unireg_sortorder(order, &length)) ||
-        (from->found_records = filesort(thd, from, sortorder, length, 
-					(SQL_SELECT *) 0, HA_POS_ERROR,
-					&examined_rows))
+        (from->sort.found_records = filesort(thd, from, sortorder, length, 
+                                             (SQL_SELECT *) 0, HA_POS_ERROR,
+                                             &examined_rows))
         == HA_POS_ERROR)
       goto err;
   };
diff --git a/sql/sql_update.cc b/sql/sql_update.cc
index 3e72f79da62..50fdfac7087 100644
--- a/sql/sql_update.cc
+++ b/sql/sql_update.cc
@@ -201,14 +201,14 @@ int mysql_update(THD *thd,
       bzero((char*) &tables,sizeof(tables));
       tables.table = table;
 
-      table->io_cache = (IO_CACHE *) my_malloc(sizeof(IO_CACHE),
+      table->sort.io_cache = (IO_CACHE *) my_malloc(sizeof(IO_CACHE),
                                                MYF(MY_FAE | MY_ZEROFILL));
       if (setup_ref_array(thd, &thd->lex.select_lex.ref_pointer_array,
 			order_num)||
 	  setup_order(thd, thd->lex.select_lex.ref_pointer_array,
 		      &tables, fields, all_fields, order) ||
           !(sortorder=make_unireg_sortorder(order, &length)) ||
-          (table->found_records = filesort(thd, table, sortorder, length,
+          (table->sort.found_records = filesort(thd, table, sortorder, length,
                                            (SQL_SELECT *) 0,
 					   HA_POS_ERROR, &examined_rows))
           == HA_POS_ERROR)
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 36ee4e54e0b..71035a75084 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -3924,7 +3924,7 @@ text_literal:
 	TEXT_STRING_literal
 	{
 	  THD *thd= YYTHD;
-	  $$ = new Item_string($1.str,$1.length,thd->variables.literal_collation);
+	  $$ = new Item_string($1.str,$1.length,thd->variables.collation_connection);
 	}
 	| NCHAR_STRING
 	{ $$=  new Item_string($1.str,$1.length,national_charset_info); }
@@ -3936,7 +3936,7 @@ text_literal:
 
 text_string:
 	TEXT_STRING_literal
-	{ $$=  new String($1.str,$1.length,YYTHD->variables.literal_collation); }
+	{ $$=  new String($1.str,$1.length,YYTHD->variables.collation_connection); }
 	| HEX_NUM
 	  {
 	    Item *tmp = new Item_varbinary($1.str,$1.length);
@@ -4106,14 +4106,14 @@ TEXT_STRING_literal:
 	TEXT_STRING
 	{
 	  THD *thd= YYTHD;
-	  if (my_charset_same(thd->charset(),thd->variables.literal_collation))
+	  if (my_charset_same(thd->charset(),thd->variables.collation_connection))
 	  {
 	    $$=$1;
 	  }
 	  else
 	  {
 	    String ident;
-	    ident.copy($1.str,$1.length,thd->charset(),thd->variables.literal_collation);
+	    ident.copy($1.str,$1.length,thd->charset(),thd->variables.collation_connection);
 	    $$.str= thd->strmake(ident.ptr(),ident.length());
 	    $$.length= ident.length();
 	  }
@@ -4405,28 +4405,27 @@ option_value:
 	{
 	  THD *thd= YYTHD;
 	  LEX *lex= Lex;
-	  $2= $2 ? $2: global_system_variables.client_collation;
+	  $2= $2 ? $2: global_system_variables.collation_client;
 	  $3= $3 ? $3 : $2;
 	  if (!my_charset_same($2,$3))
 	  {
-	      net_printf(YYTHD,ER_COLLATION_CHARSET_MISMATCH,
-			 $3->name,$2->csname);
-	      YYABORT;
+	    net_printf(thd,ER_COLLATION_CHARSET_MISMATCH,$3->name,$2->csname);
+	    YYABORT;
 	  }
-	  lex->var_list.push_back(new set_var_client_collation($3,thd->db_charset,$3));
+	  lex->var_list.push_back(new set_var_collation_client($3,thd->db_charset,$3));
 	}
 	| NAMES_SYM charset_name_or_default opt_collate
 	{
+	  THD *thd= YYTHD;
 	  LEX *lex= Lex;
-	  $2= $2 ? $2 : global_system_variables.client_collation;
+	  $2= $2 ? $2 : global_system_variables.collation_client;
 	  $3= $3 ? $3 : $2;
 	  if (!my_charset_same($2,$3))
 	  {
-	      net_printf(YYTHD,ER_COLLATION_CHARSET_MISMATCH,
-			 $3->name,$2->csname);
-	      YYABORT;
+	    net_printf(thd,ER_COLLATION_CHARSET_MISMATCH,$3->name,$2->csname);
+	    YYABORT;
 	  }
-	  lex->var_list.push_back(new set_var_client_collation($3,$3,$3));
+	  lex->var_list.push_back(new set_var_collation_client($3,$3,$3));
 	}
 	| PASSWORD equal text_or_password
 	  {
@@ -5014,5 +5013,5 @@ subselect_end:
 	')'
 	{
 	  LEX *lex=Lex;
-	  lex->current_select = lex->current_select->outer_select();
+	  lex->current_select = lex->current_select->return_after_parsing();
 	};
diff --git a/sql/structs.h b/sql/structs.h
index 77c852673d5..05ebdba7a37 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -104,6 +104,7 @@ typedef struct st_read_record {			/* Parameter to read_record */
   uint index;
   byte *ref_pos;				/* pointer to form->refpos */
   byte *record;
+  byte *rec_buf;                /* to read field values  after filesort */
   byte	*cache,*cache_pos,*cache_end,*read_positions;
   IO_CACHE *io_cache;
   bool print_error, ignore_not_found_rows;
diff --git a/sql/table.h b/sql/table.h
index 33e2db98d5a..55bc48db604 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -44,6 +44,17 @@ typedef struct st_grant_info
 
 enum tmp_table_type {NO_TMP_TABLE=0, TMP_TABLE=1, TRANSACTIONAL_TMP_TABLE=2};
 
+typedef struct st_filesort_info
+{
+  IO_CACHE *io_cache;           /* If sorted through filebyte                */
+  byte     *addon_buf;          /* Pointer to a buffer if sorted with fields */
+  uint      addon_length;       /* Length of the buffer                      */
+  struct st_sort_addon_field *addon_field;     /* Pointer to the fields info */
+  void    (*unpack)(struct st_sort_addon_field *, byte *); /* To unpack back */
+  byte     *record_pointers;    /* If sorted in memory                       */
+  ha_rows   found_records;      /* How many records in sort                  */
+} FILESORT_INFO;
+
 /* Table cache entry struct */
 
 class Field_timestamp;
@@ -120,9 +131,7 @@ struct st_table {
   table_map	map;                    /* ID bit of table (1,2,4,8,16...) */
   ulong		version,flush_version;
   uchar		*null_flags;
-  IO_CACHE	*io_cache;		/* If sorted trough filebyte */
-  byte		*record_pointers;	/* If sorted in memory */
-  ha_rows	found_records;		/* How many records in sort */
+  FILESORT_INFO sort;
   ORDER		*group;
   ha_rows	quick_rows[MAX_KEY];
   uint		quick_key_parts[MAX_KEY];
diff --git a/sql/uniques.cc b/sql/uniques.cc
index ed256a4b791..c6fb0f25643 100644
--- a/sql/uniques.cc
+++ b/sql/uniques.cc
@@ -95,12 +95,12 @@ bool Unique::flush()
 bool Unique::get(TABLE *table)
 {
   SORTPARAM sort_param;
-  table->found_records=elements+tree.elements_in_tree;
+  table->sort.found_records=elements+tree.elements_in_tree;
 
   if (my_b_tell(&file) == 0)
   {
     /* Whole tree is in memory;  Don't use disk if you don't need to */
-    if ((record_pointers=table->record_pointers= (byte*)
+    if ((record_pointers=table->sort.record_pointers= (byte*)
 	 my_malloc(tree.size_of_element * tree.elements_in_tree, MYF(0))))
     {
       (void) tree_walk(&tree, (tree_walk_action) unique_write_to_ptrs,
@@ -112,7 +112,7 @@ bool Unique::get(TABLE *table)
   if (flush())
     return 1;
 
-  IO_CACHE *outfile=table->io_cache;
+  IO_CACHE *outfile=table->sort.io_cache;
   BUFFPEK *file_ptr= (BUFFPEK*) file_ptrs.buffer;
   uint maxbuffer= file_ptrs.elements - 1;
   uchar *sort_buffer;
@@ -120,8 +120,8 @@ bool Unique::get(TABLE *table)
   bool error=1;
 
       /* Open cached file if it isn't open */
-  outfile=table->io_cache=(IO_CACHE*) my_malloc(sizeof(IO_CACHE),
-						MYF(MY_ZEROFILL));
+  outfile=table->sort.io_cache=(IO_CACHE*) my_malloc(sizeof(IO_CACHE), 
+                                MYF(MY_ZEROFILL));
 
   if (!outfile || ! my_b_inited(outfile) &&
       open_cached_file(outfile,mysql_tmpdir,TEMP_PREFIX,READ_RECORD_BUFFER,