Merge neptunus.(none):/home/msvensson/mysql/mysql-5.1

into  neptunus.(none):/home/msvensson/mysql/mysql-5.1-new-maint


BitKeeper/etc/collapsed:
  auto-union
configure.in:
  Auto merged
mysql-test/Makefile.am:
  Auto merged
mysql-test/extra/rpl_tests/rpl_deadlock.test:
  Auto merged
mysql-test/mysql-test-run-shell.sh:
  Auto merged
mysql-test/r/func_time.result:
  Auto merged
mysql-test/r/ps.result:
  Auto merged
mysql-test/r/windows.result:
  Auto merged
mysql-test/t/disabled.def:
  Auto merged
mysql-test/t/func_time.test:
  Auto merged
mysql-test/t/ps.test:
  Auto merged
mysql-test/t/windows.test:
  Auto merged
sql/item.cc:
  Auto merged
sql/item_timefunc.cc:
  Auto merged
sql/mysqld.cc:
  Auto merged
sql-common/my_time.c:
  Auto merged
sql/sql_parse.cc:
  Auto merged
BitKeeper/deleted/.del-ps_not_windows.result:
  Delete: mysql-test/r/ps_not_windows.result
BitKeeper/deleted/.del-ps_not_windows.test:
  Delete: mysql-test/t/ps_not_windows.test
mysql-test/mysql-test-run.pl:
  Manual merge
mysql-test/r/rpl_deadlock_innodb.result:
  Manual merge

34 files changed, 2590 insertions, 2084 deletions

diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index 1c2944de530..4cdc4c01c4e 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -54,6 +54,7 @@ ADD_EXECUTABLE(mysqld ../sql-common/client.c derror.cc des_key_file.cc
                event_queue.cc event_db_repository.cc 
                sql_tablespace.cc events.cc ../sql-common/my_user.c 
                partition_info.cc rpl_utility.cc rpl_injector.cc sql_locale.cc
+               rpl_rli.cc rpl_mi.cc
                ${PROJECT_SOURCE_DIR}/sql/sql_yacc.cc
   			   ${PROJECT_SOURCE_DIR}/sql/sql_yacc.h
 			   ${PROJECT_SOURCE_DIR}/include/mysqld_error.h
diff --git a/sql/Makefile.am b/sql/Makefile.am
index 9a0303a433f..ac2f4836c15 100644
--- a/sql/Makefile.am
+++ b/sql/Makefile.am
@@ -54,7 +54,7 @@ noinst_HEADERS =	item.h item_func.h item_sum.h item_cmpfunc.h \
 			ha_ndbcluster.h ha_ndbcluster_binlog.h \
 			ha_ndbcluster_tables.h \
 			opt_range.h protocol.h rpl_tblmap.h rpl_utility.h \
-			log.h sql_show.h rpl_rli.h \
+			log.h sql_show.h rpl_rli.h rpl_mi.h \
 			sql_select.h structs.h table.h sql_udf.h hash_filo.h \
 			lex.h lex_symbol.h sql_acl.h sql_crypt.h  \
 			log_event.h sql_repl.h slave.h rpl_filter.h \
@@ -94,7 +94,7 @@ mysqld_SOURCES =	sql_lex.cc sql_handler.cc sql_partition.cc \
 			sql_load.cc mf_iocache.cc field_conv.cc sql_show.cc \
 			sql_udf.cc sql_analyse.cc sql_analyse.h sql_cache.cc \
 			slave.cc sql_repl.cc rpl_filter.cc rpl_tblmap.cc \
-			rpl_utility.cc rpl_injector.cc \
+			rpl_utility.cc rpl_injector.cc rpl_rli.cc rpl_mi.cc \
                         sql_union.cc sql_derived.cc \
 			client.c sql_client.cc mini_client_errors.c pack.c\
 			stacktrace.c repl_failsafe.h repl_failsafe.cc \
diff --git a/sql/field.cc b/sql/field.cc
index 122a44305f2..1e42a53e45a 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1427,6 +1427,7 @@ Field_str::Field_str(char *ptr_arg,uint32 len_arg, uchar *null_ptr_arg,
   field_charset=charset;
   if (charset->state & MY_CS_BINSORT)
     flags|=BINARY_FLAG;
+  field_derivation= DERIVATION_IMPLICIT;
 }
 
 
@@ -5921,38 +5922,149 @@ void Field_datetime::sql_type(String &res) const
 ** A string may be varchar or binary
 ****************************************************************************/
 
+/*
+  Report "not well formed" or "cannot convert" error
+  after storing a character string info a field.
+
+  SYNOPSIS
+    check_string_copy_error()
+    field                    - Field
+    well_formed_error_pos    - where not well formed data was first met
+    cannot_convert_error_pos - where a not-convertable character was first met
+    end                      - end of the string
+
+  NOTES
+    As of version 5.0 both cases return the same error:
+  
+      "Invalid string value: 'xxx' for column 't' at row 1"
+  
+  Future versions will possibly introduce a new error message:
+
+      "Cannot convert character string: 'xxx' for column 't' at row 1"
+
+  RETURN
+    FALSE - If errors didn't happen
+    TRUE  - If an error happened
+*/
+
+static bool
+check_string_copy_error(Field_str *field,
+                        const char *well_formed_error_pos,
+                        const char *cannot_convert_error_pos,
+                        const char *end)
+{
+  const char *pos, *end_orig;
+  char tmp[64], *t;
+  
+  if (!(pos= well_formed_error_pos) &&
+      !(pos= cannot_convert_error_pos))
+    return FALSE;
+
+  end_orig= end;
+  set_if_smaller(end, pos + 6);
+
+  for (t= tmp; pos < end; pos++)
+  {
+    if (((unsigned char) *pos) >= 0x20 &&
+        ((unsigned char) *pos) <= 0x7F)
+    {
+      *t++= *pos;
+    }
+    else
+    {
+      *t++= '\\';
+      *t++= 'x';
+      *t++= _dig_vec_upper[((unsigned char) *pos) >> 4];
+      *t++= _dig_vec_upper[((unsigned char) *pos) & 15];
+    }
+  }
+  if (end_orig > end)
+  {
+    *t++= '.';
+    *t++= '.';
+    *t++= '.';
+  }
+  *t= '\0';
+  push_warning_printf(field->table->in_use, 
+                      field->table->in_use->abort_on_warning ?
+                      MYSQL_ERROR::WARN_LEVEL_ERROR :
+                      MYSQL_ERROR::WARN_LEVEL_WARN,
+                      ER_TRUNCATED_WRONG_VALUE_FOR_FIELD, 
+                      ER(ER_TRUNCATED_WRONG_VALUE_FOR_FIELD),
+                      "string", tmp, field->field_name,
+                      (ulong) field->table->in_use->row_count);
+  return TRUE;
+}
+
+
+
+/*
+  Send a truncation warning or a truncation error
+  after storing a too long character string info a field.
+
+  SYNOPSIS
+    report_data_too_long()
+    field                    - Field
+
+  RETURN
+    N/A
+*/
+
+inline void
+report_data_too_long(Field_str *field)
+{
+  if (field->table->in_use->abort_on_warning)
+    field->set_warning(MYSQL_ERROR::WARN_LEVEL_ERROR, ER_DATA_TOO_LONG, 1);
+  else
+    field->set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED, 1);
+}
+
+
+/*
+  Test if the given string contains important data:
+  not spaces for character string,
+  or any data for binary string.
+
+  SYNOPSIS
+    test_if_important_data()
+    cs          Character set
+    str         String to test
+    strend      String end
+
+  RETURN
+    FALSE - If string does not have important data
+    TRUE  - If string has some important data
+*/
+
+static bool
+test_if_important_data(CHARSET_INFO *cs, const char *str, const char *strend)
+{
+  if (cs != &my_charset_bin)
+    str+= cs->cset->scan(cs, str, strend, MY_SEQ_SPACES);
+  return (str < strend);
+}
+
+
 	/* Copy a string and fill with space */
 
 int Field_string::store(const char *from,uint length,CHARSET_INFO *cs)
 {
   ASSERT_COLUMN_MARKED_FOR_WRITE;
-  int error= 0, well_formed_error;
-  uint32 not_used;
-  char buff[STRING_BUFFER_USUAL_SIZE];
-  String tmpstr(buff,sizeof(buff), &my_charset_bin);
   uint copy_length;
+  const char *well_formed_error_pos;
+  const char *cannot_convert_error_pos;
+  const char *from_end_pos;
 
   /* See the comment for Field_long::store(long long) */
   DBUG_ASSERT(table->in_use == current_thd);
 
-  /* Convert character set if necessary */
-  if (String::needs_conversion(length, cs, field_charset, &not_used))
-  {
-    uint conv_errors;
-    tmpstr.copy(from, length, cs, field_charset, &conv_errors);
-    from= tmpstr.ptr();
-    length= tmpstr.length();
-    if (conv_errors)
-      error= 2;
-  }
-
-  /* Make sure we don't break a multibyte sequence or copy malformed data. */
-  copy_length= field_charset->cset->well_formed_len(field_charset,
-                                                    from,from+length,
-                                                    field_length/
-                                                    field_charset->mbmaxlen,
-                                                    &well_formed_error);
-  memmove(ptr, from, copy_length);
+  copy_length= well_formed_copy_nchars(field_charset,
+                                       ptr, field_length,
+                                       cs, from, length,
+                                       field_length / field_charset->mbmaxlen,
+                                       &well_formed_error_pos,
+                                       &cannot_convert_error_pos,
+                                       &from_end_pos);
 
   /* Append spaces if the string was shorter than the field. */
   if (copy_length < field_length)
@@ -5960,32 +6072,23 @@ int Field_string::store(const char *from,uint length,CHARSET_INFO *cs)
                               field_length-copy_length,
                               field_charset->pad_char);
 
+  if (check_string_copy_error(this, well_formed_error_pos,
+                              cannot_convert_error_pos, from + length))
+    return 2;
+
   /*
     Check if we lost any important data (anything in a binary string,
     or any non-space in others).
   */
-  if ((copy_length < length) && table->in_use->count_cuted_fields)
+  if ((from_end_pos < from + length) && table->in_use->count_cuted_fields)
   {
-    if (binary())
-      error= 2;
-    else
+    if (test_if_important_data(field_charset, from_end_pos, from + length))
     {
-      const char *end=from+length;
-      from+= copy_length;
-      from+= field_charset->cset->scan(field_charset, from, end,
-                                       MY_SEQ_SPACES);
-      if (from != end)
-        error= 2;
+      report_data_too_long(this);
+      return 2;
     }
   }
-  if (error)
-  {
-    if (table->in_use->abort_on_warning)
-      set_warning(MYSQL_ERROR::WARN_LEVEL_ERROR, ER_DATA_TOO_LONG, 1);
-    else
-      set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED, 1);
-  }
-  return error;
+  return 0;
 }
 
 
@@ -6343,58 +6446,35 @@ Field *Field_string::new_field(MEM_ROOT *root, struct st_table *new_table,
 int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 {
   ASSERT_COLUMN_MARKED_FOR_WRITE;
-  uint32 not_used, copy_length;
-  char buff[STRING_BUFFER_USUAL_SIZE];
-  String tmpstr(buff,sizeof(buff), &my_charset_bin);
-  int error_code= 0, well_formed_error;
-  enum MYSQL_ERROR::enum_warning_level level= MYSQL_ERROR::WARN_LEVEL_WARN;
+  uint copy_length;
+  const char *well_formed_error_pos;
+  const char *cannot_convert_error_pos;
+  const char *from_end_pos;
+
+  copy_length= well_formed_copy_nchars(field_charset,
+                                       ptr + length_bytes, field_length,
+                                       cs, from, length,
+                                       field_length / field_charset->mbmaxlen,
+                                       &well_formed_error_pos,
+                                       &cannot_convert_error_pos,
+                                       &from_end_pos);
 
-  /* Convert character set if necessary */
-  if (String::needs_conversion(length, cs, field_charset, &not_used))
-  { 
-    uint conv_errors;
-    tmpstr.copy(from, length, cs, field_charset, &conv_errors);
-    from= tmpstr.ptr();
-    length=  tmpstr.length();
-    if (conv_errors)
-      error_code= WARN_DATA_TRUNCATED;
-  }
-  /* 
-    Make sure we don't break a multibyte sequence
-    as well as don't copy a malformed data.
-  */
-  copy_length= field_charset->cset->well_formed_len(field_charset,
-						    from,from+length,
-						    field_length/
-						    field_charset->mbmaxlen,
-                                                    &well_formed_error);
-  memmove(ptr + length_bytes, from, copy_length);
   if (length_bytes == 1)
     *ptr= (uchar) copy_length;
   else
     int2store(ptr, copy_length);
 
+  if (check_string_copy_error(this, well_formed_error_pos,
+                              cannot_convert_error_pos, from + length))
+    return 2;
+
   // Check if we lost something other than just trailing spaces
-  if ((copy_length < length) && table->in_use->count_cuted_fields &&
-      !error_code)
-  {
-    if (!binary())
-    {
-      const char *end= from + length;
-      from+= copy_length;
-      from+= field_charset->cset->scan(field_charset, from, end, MY_SEQ_SPACES);
-      /* If we lost only spaces then produce a NOTE, not a WARNING */
-      if (from == end)
-        level= MYSQL_ERROR::WARN_LEVEL_NOTE;
-    }
-    error_code= WARN_DATA_TRUNCATED;
-  }
-  if (error_code)
+  if ((from_end_pos < from + length) && table->in_use->count_cuted_fields)
   {
-    if (level == MYSQL_ERROR::WARN_LEVEL_WARN &&
-        table->in_use->abort_on_warning)
-      error_code= ER_DATA_TOO_LONG;
-    set_warning(level, error_code, 1);
+    if (test_if_important_data(field_charset, from_end_pos, from + length))
+      report_data_too_long(this);
+    else /* If we lost only spaces then produce a NOTE, not a WARNING */
+      set_warning(MYSQL_ERROR::WARN_LEVEL_NOTE, WARN_DATA_TRUNCATED, 1);
     return 2;
   }
   return 0;
@@ -7012,68 +7092,70 @@ void Field_blob::put_length(char *pos, uint32 length)
 int Field_blob::store(const char *from,uint length,CHARSET_INFO *cs)
 {
   ASSERT_COLUMN_MARKED_FOR_WRITE;
-  int error= 0, well_formed_error;
+  uint copy_length, new_length;
+  const char *well_formed_error_pos;
+  const char *cannot_convert_error_pos;
+  const char *from_end_pos, *tmp;
+  char buff[STRING_BUFFER_USUAL_SIZE];
+  String tmpstr(buff,sizeof(buff), &my_charset_bin);
+
   if (!length)
   {
     bzero(ptr,Field_blob::pack_length());
+    return 0;
   }
-  else
-  {
-    bool was_conversion;
-    char buff[STRING_BUFFER_USUAL_SIZE];
-    String tmpstr(buff,sizeof(buff), &my_charset_bin);
-    uint copy_length;
-    uint32 not_used;
 
-    /* Convert character set if necessary */
-    if ((was_conversion= String::needs_conversion(length, cs, field_charset,
-						  &not_used)))
-    { 
-      uint conv_errors;
-      if (tmpstr.copy(from, length, cs, field_charset, &conv_errors))
-      {
-        /* Fatal OOM error */
-        bzero(ptr,Field_blob::pack_length());
-        return -1;
-      }
-      from= tmpstr.ptr();
-      length=  tmpstr.length();
-      if (conv_errors)
-        error= 2;
-    }
-    
-    copy_length= max_data_length();
-    /*
-      copy_length is OK as last argument to well_formed_len as this is never
-      used to limit the length of the data. The cut of long data is done with
-      the 'min()' call below.
-    */
-    copy_length= field_charset->cset->well_formed_len(field_charset,
-                                                      from,from +
-                                                      min(length, copy_length),
-                                                      copy_length,
-                                                      &well_formed_error);
-    if (copy_length < length)
-      error= 2;
-    Field_blob::store_length(copy_length);
-    if (was_conversion || table->copy_blobs || copy_length <= MAX_FIELD_WIDTH)
-    {						// Must make a copy
-      if (from != value.ptr())			// For valgrind
-      {
-	value.copy(from,copy_length,charset());
-	from=value.ptr();
-      }
+  if (from == value.ptr())
+  {
+    uint32 dummy_offset;
+    if (!String::needs_conversion(length, cs, field_charset, &dummy_offset))
+    {
+      Field_blob::store_length(length);
+      bmove(ptr+packlength,(char*) &from,sizeof(char*));
+      return 0;
     }
-    bmove(ptr+packlength,(char*) &from,sizeof(char*));
+    if (tmpstr.copy(from, length, cs))
+      goto oom_error;
+    from= tmpstr.ptr();
   }
-  if (error)
+
+  new_length= min(max_data_length(), field_charset->mbmaxlen * length);
+  if (value.alloc(new_length))
+    goto oom_error;
+
+  /*
+    "length" is OK as "nchars" argument to well_formed_copy_nchars as this
+    is never used to limit the length of the data. The cut of long data
+    is done with the new_length value.
+  */
+  copy_length= well_formed_copy_nchars(field_charset,
+                                       (char*) value.ptr(), new_length,
+                                       cs, from, length,
+                                       length,
+                                       &well_formed_error_pos,
+                                       &cannot_convert_error_pos,
+                                       &from_end_pos);
+
+  Field_blob::store_length(copy_length);
+  tmp= value.ptr();
+  bmove(ptr+packlength,(char*) &tmp,sizeof(char*));
+
+  if (check_string_copy_error(this, well_formed_error_pos,
+                              cannot_convert_error_pos, from + length))
+    return 2;
+
+  if (copy_length < length)
   {
-    if (table->in_use->abort_on_warning)
-      set_warning(MYSQL_ERROR::WARN_LEVEL_ERROR, ER_DATA_TOO_LONG, 1);
-    else
-      set_warning(MYSQL_ERROR::WARN_LEVEL_WARN, WARN_DATA_TRUNCATED, 1);
+    report_data_too_long(this);
+    return 2;
   }
+
   return 0;
+
+oom_error:
+  /* Fatal OOM error */
+  bzero(ptr,Field_blob::pack_length());
+  return -1; 
 }
 
 
diff --git a/sql/field.h b/sql/field.h
index 9b81931d416..f0cd9cc6f03 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -351,6 +351,9 @@ public:
   virtual CHARSET_INFO *sort_charset(void) const { return charset(); }
   virtual bool has_charset(void) const { return FALSE; }
   virtual void set_charset(CHARSET_INFO *charset) { }
+  virtual enum Derivation derivation(void) const
+  { return DERIVATION_IMPLICIT; }
+  virtual void set_derivation(enum Derivation derivation) { }
   bool set_warning(MYSQL_ERROR::enum_warning_level, unsigned int code,
                    int cuted_increment);
   bool check_int(const char *str, int length, const char *int_end,
@@ -446,6 +449,7 @@ public:
 class Field_str :public Field {
 protected:
   CHARSET_INFO *field_charset;
+  enum Derivation field_derivation;
 public:
   Field_str(char *ptr_arg,uint32 len_arg, uchar *null_ptr_arg,
 	    uchar null_bit_arg, utype unireg_check_arg,
@@ -459,6 +463,9 @@ public:
   uint size_of() const { return sizeof(*this); }
   CHARSET_INFO *charset(void) const { return field_charset; }
   void set_charset(CHARSET_INFO *charset) { field_charset=charset; }
+  enum Derivation derivation(void) const { return field_derivation; }
+  virtual void set_derivation(enum Derivation derivation_arg)
+  { field_derivation= derivation_arg; }
   bool binary() const { return field_charset == &my_charset_bin; }
   uint32 max_length() { return field_length; }
   friend class create_field;
diff --git a/sql/ha_ndbcluster.cc b/sql/ha_ndbcluster.cc
index 9df2171d85c..76999896e3c 100644
--- a/sql/ha_ndbcluster.cc
+++ b/sql/ha_ndbcluster.cc
@@ -413,7 +413,7 @@ Thd_ndb::get_open_table(THD *thd, const void *key)
     thd_ndb_share->stat.no_uncommitted_rows_count= 0;
     thd_ndb_share->stat.records= ~(ha_rows)0;
   }
-  DBUG_PRINT("exit", ("thd_ndb_share: 0x%x  key: 0x%x", thd_ndb_share, key));
+  DBUG_PRINT("exit", ("thd_ndb_share: %p  key: %p", thd_ndb_share, key));
   DBUG_RETURN(thd_ndb_share);
 }
 
@@ -1171,7 +1171,7 @@ int ha_ndbcluster::add_index_handle(THD *thd, NDBDICT *dict, KEY *key_info,
       index= dict->getIndexGlobal(index_name, *m_table);
       if (!index)
         ERR_RETURN(dict->getNdbError());
-      DBUG_PRINT("info", ("index: 0x%x  id: %d  version: %d.%d  status: %d",
+      DBUG_PRINT("info", ("index: %p  id: %d  version: %d.%d  status: %d",
                           index,
                           index->getObjectId(),
                           index->getObjectVersion() & 0xFFFFFF,
@@ -1215,7 +1215,7 @@ int ha_ndbcluster::add_index_handle(THD *thd, NDBDICT *dict, KEY *key_info,
       index= dict->getIndexGlobal(unique_index_name, *m_table);
       if (!index)
         ERR_RETURN(dict->getNdbError());
-      DBUG_PRINT("info", ("index: 0x%x  id: %d  version: %d.%d  status: %d",
+      DBUG_PRINT("info", ("index: %p  id: %d  version: %d.%d  status: %d",
                           index,
                           index->getObjectId(),
                           index->getObjectVersion() & 0xFFFFFF,
@@ -2072,7 +2072,7 @@ inline int ha_ndbcluster::fetch_next(NdbScanOperation* cursor)
         all pending update or delete operations should 
         be sent to NDB
       */
-      DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending));    
+      DBUG_PRINT("info", ("ops_pending: %llu", m_ops_pending));
       if (m_ops_pending)
       {
         if (m_transaction_on)
@@ -3480,7 +3480,7 @@ int ha_ndbcluster::close_scan()
       Take over any pending transactions to the 
       deleteing/updating transaction before closing the scan    
     */
-    DBUG_PRINT("info", ("ops_pending: %d", m_ops_pending));    
+    DBUG_PRINT("info", ("ops_pending: %llu", m_ops_pending));
     if (execute_no_commit(this,trans,false) != 0) {
       no_uncommitted_rows_execute_failure();
       DBUG_RETURN(ndb_err(trans));
@@ -4286,7 +4286,7 @@ static int ndbcluster_commit(handlerton *hton, THD *thd, bool all)
   while ((share= it++))
   {
     pthread_mutex_lock(&share->mutex);
-    DBUG_PRINT("info", ("Invalidate commit_count for %s, share->commit_count: %d ",
+    DBUG_PRINT("info", ("Invalidate commit_count for %s, commit_count: %llu ",
 			share->key, share->commit_count));
     share->commit_count= 0;
     share->commit_count_lock++;
@@ -4691,7 +4691,7 @@ int ha_ndbcluster::create(const char *name,
     DBUG_RETURN(2);
   }
 
-  DBUG_PRINT("info", ("setFrm data=%lx  len=%d", pack_data, pack_length));
+  DBUG_PRINT("info", ("setFrm data=%p  len=%d", pack_data, pack_length));
   tab.setFrm(pack_data, pack_length);      
   my_free((char*)data, MYF(0));
   my_free((char*)pack_data, MYF(0));
@@ -5103,8 +5103,8 @@ int ha_ndbcluster::add_index(TABLE *table_arg,
                              KEY *key_info, uint num_of_keys)
 {
   DBUG_ENTER("ha_ndbcluster::add_index");
-  DBUG_PRINT("info", ("ha_ndbcluster::add_index to table %s", 
-                      table_arg->s->table_name));
+  DBUG_PRINT("info", ("ha_ndbcluster::add_index to table %s",
+                      table_arg->s->table_name.str));
   int error= 0;
   uint idx;
 
@@ -6103,9 +6103,7 @@ static void ndbcluster_drop_database(handlerton *hton, char *path)
 #endif
   DBUG_VOID_RETURN;
 }
-/*
-  find all tables in ndb and discover those needed
-*/
+
 int ndb_create_table_from_engine(THD *thd, const char *db,
                                  const char *table_name)
 {
@@ -6118,6 +6116,9 @@ int ndb_create_table_from_engine(THD *thd, const char *db,
   return res;
 }
 
+/*
+  find all tables in ndb and discover those needed
+*/
 int ndbcluster_find_all_files(THD *thd)
 {
   DBUG_ENTER("ndbcluster_find_all_files");
@@ -7162,31 +7163,51 @@ static byte *ndbcluster_get_key(NDB_SHARE *share,uint *length,
   return (byte*) share->key;
 }
 
+
 #ifndef DBUG_OFF
-static void dbug_print_open_tables()
+
+static void print_share(const char* where, NDB_SHARE* share)
 {
-  DBUG_ENTER("dbug_print_open_tables");
-  for (uint i= 0; i < ndbcluster_open_tables.records; i++)
-  {
-    NDB_SHARE *share= (NDB_SHARE*) hash_element(&ndbcluster_open_tables, i);
-    DBUG_PRINT("share",
-               ("[%d] 0x%lx key: %s  key_length: %d",
-                i, share, share->key, share->key_length));
-    DBUG_PRINT("share",
-               ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
-                share->db, share->table_name,
-                share->use_count, share->commit_count));
+  fprintf(DBUG_FILE,
+          "%s %s.%s: use_count: %u, commit_count: %llu\n",
+          where, share->db, share->table_name, share->use_count,
+          share->commit_count);
+  fprintf(DBUG_FILE,
+          "  - key: %s, key_length: %d\n",
+          share->key, share->key_length);
+
 #ifdef HAVE_NDB_BINLOG
-    if (share->table)
-      DBUG_PRINT("share",
-                 ("table->s->db.table_name: %s.%s",
-                  share->table->s->db.str, share->table->s->table_name.str));
+  if (share->table)
+    fprintf(DBUG_FILE,
+            "  - share->table: %p %s.%s\n",
+            share->table, share->table->s->db.str,
+            share->table->s->table_name.str);
 #endif
-  }
-  DBUG_VOID_RETURN;
 }
-#else
-#define dbug_print_open_tables()
+
+
+static void print_ndbcluster_open_tables()
+{
+  DBUG_LOCK_FILE;
+  fprintf(DBUG_FILE, ">ndbcluster_open_tables\n");
+  for (uint i= 0; i < ndbcluster_open_tables.records; i++)
+    print_share("",
+                (NDB_SHARE*)hash_element(&ndbcluster_open_tables, i));
+  fprintf(DBUG_FILE, "<ndbcluster_open_tables\n");
+  DBUG_UNLOCK_FILE;
+}
+
+
+#define dbug_print_open_tables()                \
+  DBUG_EXECUTE("info",                          \
+               print_ndbcluster_open_tables(););
+
+#define dbug_print_share(t, s)                  \
+  DBUG_LOCK_FILE;                               \
+  DBUG_EXECUTE("info",                          \
+               print_share((t), (s)););         \
+  DBUG_UNLOCK_FILE;
+
 #endif
 
 #ifdef HAVE_NDB_BINLOG
@@ -7331,19 +7352,9 @@ static int rename_share(NDB_SHARE *share, const char *new_key)
   share->table_name= share->db + strlen(share->db) + 1;
   ha_ndbcluster::set_tabname(new_key, share->table_name);
 
-  DBUG_PRINT("rename_share",
-             ("0x%lx key: %s  key_length: %d",
-              share, share->key, share->key_length));
-  DBUG_PRINT("rename_share",
-             ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
-              share->db, share->table_name,
-              share->use_count, share->commit_count));
+  dbug_print_share("rename_share:", share);
   if (share->table)
   {
-    DBUG_PRINT("rename_share",
-               ("table->s->db.table_name: %s.%s",
-                share->table->s->db.str, share->table->s->table_name.str));
-
     if (share->op == 0)
     {
       share->table->s->db.str= share->db;
@@ -7371,14 +7382,7 @@ NDB_SHARE *ndbcluster_get_share(NDB_SHARE *share)
   share->use_count++;
 
   dbug_print_open_tables();
-
-  DBUG_PRINT("get_share",
-             ("0x%lx key: %s  key_length: %d",
-              share, share->key, share->key_length));
-  DBUG_PRINT("get_share",
-             ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
-              share->db, share->table_name,
-              share->use_count, share->commit_count));
+  dbug_print_share("ndbcluster_get_share:", share);
   pthread_mutex_unlock(&ndbcluster_mutex);
   return share;
 }
@@ -7469,14 +7473,7 @@ NDB_SHARE *ndbcluster_get_share(const char *key, TABLE *table,
   share->use_count++;
 
   dbug_print_open_tables();
-
-  DBUG_PRINT("info",
-             ("0x%lx key: %s  key_length: %d  key: %s",
-              share, share->key, share->key_length, key));
-  DBUG_PRINT("info",
-             ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
-              share->db, share->table_name,
-              share->use_count, share->commit_count));
+  dbug_print_share("ndbcluster_get_share:", share);
   if (!have_lock)
     pthread_mutex_unlock(&ndbcluster_mutex);
   DBUG_RETURN(share);
@@ -7486,13 +7483,7 @@ NDB_SHARE *ndbcluster_get_share(const char *key, TABLE *table,
 void ndbcluster_real_free_share(NDB_SHARE **share)
 {
   DBUG_ENTER("ndbcluster_real_free_share");
-  DBUG_PRINT("real_free_share",
-             ("0x%lx key: %s  key_length: %d",
-              (*share), (*share)->key, (*share)->key_length));
-  DBUG_PRINT("real_free_share",
-             ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
-              (*share)->db, (*share)->table_name,
-              (*share)->use_count, (*share)->commit_count));
+  dbug_print_share("ndbcluster_real_free_share:", *share);
 
   hash_delete(&ndbcluster_open_tables, (byte*) *share);
   thr_lock_delete(&(*share)->lock);
@@ -7521,12 +7512,7 @@ void ndbcluster_real_free_share(NDB_SHARE **share)
   DBUG_VOID_RETURN;
 }
 
-/*
-  decrease refcount of share
-  calls real_free_share when refcount reaches 0
 
-  have_lock == TRUE, pthread_mutex_lock(&ndbcluster_mutex) already taken
-*/
 void ndbcluster_free_share(NDB_SHARE **share, bool have_lock)
 {
   if (!have_lock)
@@ -7540,13 +7526,7 @@ void ndbcluster_free_share(NDB_SHARE **share, bool have_lock)
   else
   {
     dbug_print_open_tables();
-    DBUG_PRINT("free_share",
-               ("0x%lx key: %s  key_length: %d",
-                *share, (*share)->key, (*share)->key_length));
-    DBUG_PRINT("free_share",
-               ("db.tablename: %s.%s  use_count: %d  commit_count: %d",
-                (*share)->db, (*share)->table_name,
-                (*share)->use_count, (*share)->commit_count));
+    dbug_print_share("ndbcluster_free_share:", *share);
   }
   if (!have_lock)
     pthread_mutex_unlock(&ndbcluster_mutex);
@@ -8176,27 +8156,25 @@ ha_ndbcluster::update_table_comment(
 pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
 {
   THD *thd; /* needs to be first for thread_stack */
-  Ndb* ndb;
   struct timespec abstime;
   List<NDB_SHARE> util_open_tables;
+  Thd_ndb *thd_ndb;
 
   my_thread_init();
   DBUG_ENTER("ndb_util_thread");
-  DBUG_PRINT("enter", ("ndb_cache_check_time: %d", ndb_cache_check_time));
+  DBUG_PRINT("enter", ("ndb_cache_check_time: %lu", ndb_cache_check_time));
 
   thd= new THD; /* note that contructor of THD uses DBUG_ */
   THD_CHECK_SENTRY(thd);
-  ndb= new Ndb(g_ndb_cluster_connection, "");
 
   pthread_detach_this_thread();
   ndb_util_thread= pthread_self();
 
   thd->thread_stack= (char*)&thd; /* remember where our stack is */
-  if (thd->store_globals() || (ndb->init() != 0))
+  if (thd->store_globals())
   {
     thd->cleanup();
     delete thd;
-    delete ndb;
     DBUG_RETURN(NULL);
   }
   thd->init_for_queries();
@@ -8238,16 +8216,14 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
   }
   pthread_mutex_unlock(&LOCK_ndb_util_thread);
 
+  /* Get thd_ndb for this thread */
+  if (!(thd_ndb= ha_ndbcluster::seize_thd_ndb()))
   {
-    Thd_ndb *thd_ndb;
-    if (!(thd_ndb= ha_ndbcluster::seize_thd_ndb()))
-    {
-      sql_print_error("Could not allocate Thd_ndb object");
-      goto ndb_util_thread_end;
-    }
-    set_thd_ndb(thd, thd_ndb);
-    thd_ndb->options|= TNO_NO_LOG_SCHEMA_OP;
+    sql_print_error("Could not allocate Thd_ndb object");
+    goto ndb_util_thread_end;
   }
+  set_thd_ndb(thd, thd_ndb);
+  thd_ndb->options|= TNO_NO_LOG_SCHEMA_OP;
 
 #ifdef HAVE_NDB_BINLOG
   if (ndb_extra_logging && ndb_binlog_running)
@@ -8330,26 +8306,26 @@ pthread_handler_t ndb_util_thread_func(void *arg __attribute__((unused)))
       }
 #endif /* HAVE_NDB_BINLOG */
       DBUG_PRINT("ndb_util_thread",
-                 ("Fetching commit count for: %s",
-                  share->key));
+                 ("Fetching commit count for: %s", share->key));
 
-      /* Contact NDB to get commit count for table */
-      ndb->setDatabaseName(share->db);
       struct Ndb_statistics stat;
-
       uint lock;
       pthread_mutex_lock(&share->mutex);
       lock= share->commit_count_lock;
       pthread_mutex_unlock(&share->mutex);
 
       {
+        /* Contact NDB to get commit count for table */
+        Ndb* ndb= thd_ndb->ndb;
+        ndb->setDatabaseName(share->db);
         Ndb_table_guard ndbtab_g(ndb->getDictionary(), share->table_name);
         if (ndbtab_g.get_table() &&
-            ndb_get_table_statistics(NULL, false, ndb, ndbtab_g.get_table(), &stat) == 0)
+            ndb_get_table_statistics(NULL, false, ndb,
+                                     ndbtab_g.get_table(), &stat) == 0)
         {
           char buff[22], buff2[22];
           DBUG_PRINT("ndb_util_thread",
-                     ("Table: %s, commit_count: %llu, rows: %llu",
+                     ("Table: %s, commit_count: %s, rows: %s",
                       share->key,
                       llstr(stat.commit_count, buff),
                       llstr(stat.row_count, buff2)));
@@ -8401,7 +8377,6 @@ ndb_util_thread_end:
   net_end(&thd->net);
   thd->cleanup();
   delete thd;
-  delete ndb;
   DBUG_PRINT("exit", ("ndb_util_thread"));
   my_thread_end();
   pthread_exit(0);
@@ -9191,7 +9166,7 @@ void ndb_serialize_cond(const Item *item, void *arg)
           if (context->expecting(Item::INT_ITEM)) 
           {
             Item_int *int_item= (Item_int *) item;      
-            DBUG_PRINT("info", ("value %d", int_item->value));
+            DBUG_PRINT("info", ("value %llu", int_item->value));
             NDB_ITEM_QUALIFICATION q;
             q.value_type= Item::INT_ITEM;
             curr_cond->ndb_item= new Ndb_item(NDB_VALUE, q, item);
@@ -9214,7 +9189,7 @@ void ndb_serialize_cond(const Item *item, void *arg)
             context->supported= FALSE;
           break;
         case Item::REAL_ITEM:
-          DBUG_PRINT("info", ("REAL_ITEM %s"));
+          DBUG_PRINT("info", ("REAL_ITEM"));
           if (context->expecting(Item::REAL_ITEM)) 
           {
             Item_float *float_item= (Item_float *) item;      
@@ -9262,7 +9237,7 @@ void ndb_serialize_cond(const Item *item, void *arg)
             context->supported= FALSE;
           break;
         case Item::DECIMAL_ITEM:
-          DBUG_PRINT("info", ("DECIMAL_ITEM %s"));
+          DBUG_PRINT("info", ("DECIMAL_ITEM"));
           if (context->expecting(Item::DECIMAL_ITEM)) 
           {
             Item_decimal *decimal_item= (Item_decimal *) item;      
diff --git a/sql/ha_ndbcluster_binlog.cc b/sql/ha_ndbcluster_binlog.cc
index 3dfca5d1bb2..e0b7502a40a 100644
--- a/sql/ha_ndbcluster_binlog.cc
+++ b/sql/ha_ndbcluster_binlog.cc
@@ -3436,7 +3436,6 @@ pthread_handler_t ndb_binlog_thread_func(void *arg)
         global_system_variables.binlog_format == BINLOG_FORMAT_MIXED)
     {
       ndb_binlog_running= TRUE;
-      thd->current_stmt_binlog_row_based= TRUE;     // If in mixed mode
     }
     else
     {
diff --git a/sql/handler.cc b/sql/handler.cc
index 451f974a066..f16876f2ffd 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -3460,38 +3460,15 @@ bool ha_show_status(THD *thd, handlerton *db_type, enum ha_stat_type stat)
    declared static, but it works by putting it into an anonymous
    namespace. */
 namespace {
-  struct st_table_data {
-    char const *db;
-    char const *name;
-  };
-
-  static int table_name_compare(void const *a, void const *b)
-  {
-    st_table_data const *x = (st_table_data const*) a;
-    st_table_data const *y = (st_table_data const*) b;
-
-    /* Doing lexical compare in order (db,name) */
-    int const res= strcmp(x->db, y->db);
-    return res != 0 ? res : strcmp(x->name, y->name);
-  }
-
   bool check_table_binlog_row_based(THD *thd, TABLE *table)
   {
-    static st_table_data const ignore[] = {
-      { "mysql", "event" },
-      { "mysql", "general_log" },
-      { "mysql", "slow_log" }
-    };
-
-    my_size_t const ignore_size = sizeof(ignore)/sizeof(*ignore);
-    st_table_data const item = { table->s->db.str, table->s->table_name.str };
-
     if (table->s->cached_row_logging_check == -1)
-      table->s->cached_row_logging_check=
-        (table->s->tmp_table == NO_TMP_TABLE) &&
-        binlog_filter->db_ok(table->s->db.str) &&
-        bsearch(&item, ignore, ignore_size,
-                sizeof(st_table_data), table_name_compare) == NULL;
+    {
+      int const check(table->s->tmp_table == NO_TMP_TABLE &&
+                      binlog_filter->db_ok(table->s->db.str) &&
+                      strcmp("mysql", table->s->db.str) != 0);
+      table->s->cached_row_logging_check= check;
+    }
 
     DBUG_ASSERT(table->s->cached_row_logging_check == 0 ||
                 table->s->cached_row_logging_check == 1);
diff --git a/sql/item.cc b/sql/item.cc
index 8914593e9f0..0662ddefbb4 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1655,7 +1655,7 @@ void Item_field::set_field(Field *field_par)
   db_name= field_par->table->s->db.str;
   alias_name_used= field_par->table->alias_name_used;
   unsigned_flag=test(field_par->flags & UNSIGNED_FLAG);
-  collation.set(field_par->charset(), DERIVATION_IMPLICIT);
+  collation.set(field_par->charset(), field_par->derivation());
   fixed= 1;
 }
 
diff --git a/sql/item.h b/sql/item.h
index 8799fa07eb7..2c26e1c4a07 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -27,19 +27,7 @@ class Item_field;
 /*
    "Declared Type Collation"
    A combination of collation and its derivation.
-*/
 
-enum Derivation
-{
-  DERIVATION_IGNORABLE= 5,
-  DERIVATION_COERCIBLE= 4,
-  DERIVATION_SYSCONST= 3,
-  DERIVATION_IMPLICIT= 2,
-  DERIVATION_NONE= 1,
-  DERIVATION_EXPLICIT= 0
-};
-
-/*
   Flags for collation aggregation modes:
   MY_COLL_ALLOW_SUPERSET_CONV  - allow conversion to a superset
   MY_COLL_ALLOW_COERCIBLE_CONV - allow conversion of a coercible value
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index b2ff62028c1..1cb943e45c2 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -3019,6 +3019,7 @@ int dump_leaf_key(byte* key, element_count count __attribute__((unused)),
   String tmp2;
   String *result= &item->result;
   Item **arg= item->args, **arg_end= item->args + item->arg_count_field;
+  uint old_length= result->length();
 
   if (item->no_appended)
     item->no_appended= FALSE;
@@ -3053,8 +3054,22 @@ int dump_leaf_key(byte* key, element_count count __attribute__((unused)),
   /* stop if length of result more than max_length */
   if (result->length() > item->max_length)
   {
+    int well_formed_error;
+    CHARSET_INFO *cs= item->collation.collation;
+    const char *ptr= result->ptr();
+    uint add_length;
+    /*
+      It's ok to use item->result.length() as the fourth argument
+      as this is never used to limit the length of the data.
+      Cut is done with the third argument.
+    */
+    add_length= cs->cset->well_formed_len(cs,
+                                          ptr + old_length,
+                                          ptr + item->max_length,
+                                          result->length(),
+                                          &well_formed_error);
+    result->length(old_length + add_length);
     item->count_cut_values++;
-    result->length(item->max_length);
     item->warning_for_row= TRUE;
     return 1;
   }
@@ -3244,8 +3259,7 @@ bool Item_func_group_concat::add()
     we can dump the row here in case of GROUP_CONCAT(DISTINCT...)
     instead of doing tree traverse later.
   */
-  if (result.length() <= max_length &&
-      !warning_for_row &&
+  if (!warning_for_row &&
       (!tree || (el->count == 1 && distinct && !arg_count_order)))
     dump_leaf_key(table->record[0] + table->s->null_bytes, 1, this);
 
@@ -3314,7 +3328,8 @@ bool Item_func_group_concat::setup(THD *thd)
     DBUG_RETURN(TRUE);
 
   /* We'll convert all blobs to varchar fields in the temporary table */
-  tmp_table_param->convert_blob_length= max_length;
+  tmp_table_param->convert_blob_length= max_length *
+                                        collation.collation->mbmaxlen;
   /* Push all not constant fields to the list and create a temp table */
   always_null= 0;
   for (uint i= 0; i < arg_count_field; i++)
diff --git a/sql/item_xmlfunc.cc b/sql/item_xmlfunc.cc
index 44a2b690bac..21239a13735 100644
--- a/sql/item_xmlfunc.cc
+++ b/sql/item_xmlfunc.cc
@@ -532,7 +532,7 @@ public:
   longlong val_int()
   {
     Item_func *comp= (Item_func*)args[1];
-    Item_string *fake= (Item_string*)(comp->arguments()[1]);
+    Item_string *fake= (Item_string*)(comp->arguments()[0]);
     String *res= args[0]->val_nodeset(&tmp_nodeset);
     MY_XPATH_FLT *fltbeg= (MY_XPATH_FLT*) res->ptr();
     MY_XPATH_FLT *fltend= (MY_XPATH_FLT*) (res->ptr() + res->length());
@@ -884,7 +884,7 @@ static Item *eq_func(int oper, Item *a, Item *b)
     Create a comparator function for scalar arguments,
     for the given arguments and reverse operation, e.g.
 
-    A >= B  is converted into  A < B
+    A > B  is converted into  B < A
 
   RETURN
     The newly created item.
@@ -895,10 +895,10 @@ static Item *eq_func_reverse(int oper, Item *a, Item *b)
   {
     case '=': return new Item_func_eq(a, b);
     case '!': return new Item_func_ne(a, b);
-    case MY_XPATH_LEX_GE: return new Item_func_lt(a, b);
-    case MY_XPATH_LEX_LE: return new Item_func_gt(a, b);
-    case MY_XPATH_LEX_GREATER: return new Item_func_le(a, b);
-    case MY_XPATH_LEX_LESS: return new Item_func_ge(a, b);
+    case MY_XPATH_LEX_GE: return new Item_func_le(a, b);
+    case MY_XPATH_LEX_LE: return new Item_func_ge(a, b);
+    case MY_XPATH_LEX_GREATER: return new Item_func_lt(a, b);
+    case MY_XPATH_LEX_LESS: return new Item_func_gt(a, b);
   }
   return 0;
 }
@@ -951,13 +951,13 @@ static Item *create_comparator(MY_XPATH *xpath,
     {
       nodeset= (Item_nodeset_func*) a;
       scalar= b;
-      comp= eq_func(oper, scalar, fake);
+      comp= eq_func(oper, fake, scalar);
     }
     else
     {
       nodeset= (Item_nodeset_func*) b;
       scalar= a;
-      comp= eq_func_reverse(oper, scalar, fake);
+      comp= eq_func_reverse(oper, fake, scalar);
     }
     return new Item_nodeset_to_const_comparator(nodeset, comp, xpath->pxml);
   }
diff --git a/sql/log.cc b/sql/log.cc
index 83e190a5c01..620445aecfa 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -424,6 +424,18 @@ bool Log_to_csv_event_handler::
 {
   TABLE *table= general_log.table;
 
+  /*
+    "INSERT INTO general_log" can generate warning sometimes.
+    Let's reset warnings from previous queries,
+    otherwise warning list can grow too much,
+    so thd->query gets spoiled as some point in time,
+    and mysql_parse() receives a broken query.
+    QQ: this problem needs to be studied in more details.
+    Probably it's better to suppress warnings in logging INSERTs at all.
+    Comment this line and run "cast.test" to see what's happening:
+  */
+  mysql_reset_errors(table->in_use, 1);
+
   /* below should never happen */
   if (unlikely(!logger.is_log_tables_initialized))
     return FALSE;
diff --git a/sql/log_event.cc b/sql/log_event.cc
index d99fb9da8f8..f11822d8ad0 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -140,6 +140,21 @@ static void pretty_print_str(IO_CACHE* cache, char* str, int len)
 }
 #endif /* MYSQL_CLIENT */
 
+#ifdef HAVE_purify
+static void
+valgrind_check_mem(void *ptr, size_t len)
+{
+  static volatile uchar dummy;
+  for (volatile uchar *p= (uchar*) ptr ; p != (uchar*) ptr + len ; ++p)
+  {
+    int const c = *p;
+    if (c < 128)
+      dummy= c + 1;
+    else
+      dummy = c - 1;
+  }
+}
+#endif
 
 #if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
 
@@ -1398,6 +1413,7 @@ bool Query_log_event::write(IO_CACHE* file)
   
   /* Store length of status variables */
   status_vars_len= (uint) (start-start_of_status);
+  DBUG_ASSERT(status_vars_len <= MAX_SIZE_LOG_EVENT_STATUS);
   int2store(buf + Q_STATUS_VARS_LEN_OFFSET, status_vars_len);
 
   /*
@@ -5401,7 +5417,13 @@ int Rows_log_event::do_add_row_data(byte *const row_data,
   DBUG_ENTER("Rows_log_event::do_add_row_data");
   DBUG_PRINT("enter", ("row_data: 0x%lx  length: %lu", (ulong) row_data,
                        length));
+  /*
+    Don't print debug messages when running valgrind since they can
+    trigger false warnings.
+   */
+#ifndef HAVE_purify
   DBUG_DUMP("row_data", (const char*)row_data, min(length, 32));
+#endif
 
   DBUG_ASSERT(m_rows_buf <= m_rows_cur);
   DBUG_ASSERT(!m_rows_buf || m_rows_end && m_rows_buf < m_rows_end);
@@ -5445,14 +5467,13 @@ int Rows_log_event::do_add_row_data(byte *const row_data,
 
 #if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
 /*
-  Unpack a row into a record.
+  Unpack a row into table->record[0].
   
   SYNOPSIS
     unpack_row()
     rli     Relay log info
     table   Table to unpack into
     colcnt  Number of columns to read from record
-    record  Record where the data should be unpacked
     row     Packed row data
     cols    Pointer to columns data to fill in
     row_end Pointer to variable that will hold the value of the
@@ -5465,6 +5486,11 @@ int Rows_log_event::do_add_row_data(byte *const row_data,
 
   DESCRIPTION
 
+      The function will always unpack into the table->record[0]
+      record.  This is because there are too many dependencies on
+      where the various member functions of Field and subclasses
+      expect to write.
+
       The row is assumed to only consist of the fields for which the
       bitset represented by 'arr' and 'bits'; the other parts of the
       record are left alone.
@@ -5483,13 +5509,15 @@ int Rows_log_event::do_add_row_data(byte *const row_data,
  */
 static int
 unpack_row(RELAY_LOG_INFO *rli,
-           TABLE *table, uint const colcnt, byte *record,
+           TABLE *table, uint const colcnt,
            char const *row, MY_BITMAP const *cols,
            char const **row_end, ulong *master_reclength,
            MY_BITMAP* const rw_set, Log_event_type const event_type)
 {
+  byte *const record= table->record[0];
+  DBUG_ENTER("unpack_row");
   DBUG_ASSERT(record && row);
-  my_ptrdiff_t const offset= record - (byte*) table->record[0];
+  DBUG_PRINT("enter", ("row=0x%lx; table->record[0]=0x%lx", row, record));
   my_size_t master_null_bytes= table->s->null_bytes;
 
   if (colcnt != table->s->fields)
@@ -5529,9 +5557,11 @@ unpack_row(RELAY_LOG_INFO *rli,
 
     if (bitmap_is_set(cols, field_ptr -  begin_ptr))
     {
-      f->move_field_offset(offset);
+      DBUG_ASSERT(table->record[0] <= f->ptr);
+      DBUG_ASSERT(f->ptr < table->record[0] + table->s->reclength + (f->pack_length_in_rec() == 0));
+
+      DBUG_PRINT("info", ("unpacking column '%s' to 0x%lx", f->field_name, f->ptr));
       ptr= f->unpack(f->ptr, ptr);
-      f->move_field_offset(-offset);
       /* Field...::unpack() cannot return 0 */
       DBUG_ASSERT(ptr != NULL);
     }
@@ -5562,13 +5592,10 @@ unpack_row(RELAY_LOG_INFO *rli,
   for ( ; *field_ptr ; ++field_ptr)
   {
     uint32 const mask= NOT_NULL_FLAG | NO_DEFAULT_VALUE_FLAG;
+    Field *const f= *field_ptr;
 
-    DBUG_PRINT("debug", ("flags = 0x%x, mask = 0x%x, flags & mask = 0x%x",
-                         (*field_ptr)->flags, mask,
-                         (*field_ptr)->flags & mask));
-
-    if (event_type == WRITE_ROWS_EVENT &&
-        ((*field_ptr)->flags & mask) == mask)
+    DBUG_PRINT("info", ("processing column '%s' @ 0x%lx", f->field_name, f->ptr));
+    if (event_type == WRITE_ROWS_EVENT && (f->flags & mask) == mask)
     {
       slave_print_msg(ERROR_LEVEL, rli, ER_NO_DEFAULT_FOR_FIELD,
                       "Field `%s` of table `%s`.`%s` "
@@ -5578,10 +5605,10 @@ unpack_row(RELAY_LOG_INFO *rli,
       error = ER_NO_DEFAULT_FOR_FIELD;
     }
     else
-      (*field_ptr)->set_default();
+      f->set_default();
   }
 
-  return error;
+  DBUG_RETURN(error);
 }
 
 int Rows_log_event::exec_event(st_relay_log_info *rli)
@@ -5689,12 +5716,10 @@ int Rows_log_event::exec_event(st_relay_log_info *rli)
       We also invalidate the query cache for all the tables, since
       they will now be changed.
      */
-    
     TABLE_LIST *ptr;
     for (ptr= rli->tables_to_lock ; ptr ; ptr= ptr->next_global)
     {
       rli->m_table_map.set_table(ptr->table_id, ptr->table);
-      rli->touching_table(ptr->db, ptr->table_name, ptr->table_id);
     }
 #ifdef HAVE_QUERY_CACHE
     query_cache.invalidate_locked_for_write(rli->tables_to_lock);
@@ -5747,7 +5772,7 @@ int Rows_log_event::exec_event(st_relay_log_info *rli)
       if ((error= do_prepare_row(thd, rli, table, row_start, &row_end)))
         break; // We should perform the after-row operation even in
                // the case of error
-      
+
       DBUG_ASSERT(row_end != NULL); // cannot happen
       DBUG_ASSERT(row_end <= (const char*)m_rows_end);
 
@@ -5803,9 +5828,10 @@ int Rows_log_event::exec_event(st_relay_log_info *rli)
       STMT_END_F.
       For now we code, knowing that error is not skippable and so slave SQL
       thread is certainly going to stop.
+      rollback at the caller along with sbr.
     */
     thd->reset_current_stmt_binlog_row_based();
-    rli->cleanup_context(thd, 1);
+    rli->cleanup_context(thd, 0);  /* rollback at caller in step with sbr */
     thd->query_error= 1;
     DBUG_RETURN(error);
   }
@@ -6047,7 +6073,13 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len,
   DBUG_PRINT("info",("event_len=%ld, common_header_len=%d, post_header_len=%d",
                      event_len, common_header_len, post_header_len));
 
+  /*
+    Don't print debug messages when running valgrind since they can
+    trigger false warnings.
+   */
+#ifndef HAVE_purify
   DBUG_DUMP("event buffer", buf, event_len);
+#endif
 
   /* Read the post-header */
   const char *post_start= buf + common_header_len;
@@ -6230,8 +6262,7 @@ int Table_map_log_event::exec_event(st_relay_log_info *rli)
 
     /*
       We record in the slave's information that the table should be
-      locked by linking the table into the list of tables to lock, and
-      tell the RLI that we are touching a table.
+      locked by linking the table into the list of tables to lock.
     */
     table_list->next_global= table_list->next_local= rli->tables_to_lock;
     rli->tables_to_lock= table_list;
@@ -6423,17 +6454,15 @@ int Write_rows_log_event::do_after_row_operations(TABLE *table, int error)
 
 int Write_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
                                          TABLE *table,
-                                         char const *row_start,
-                                         char const **row_end)
+                                         char const *const row_start,
+                                         char const **const row_end)
 {
   DBUG_ASSERT(table != NULL);
   DBUG_ASSERT(row_start && row_end);
 
   int error;
-  error= unpack_row(rli,
-                    table, m_width, table->record[0],
-                    row_start, &m_cols, row_end, &m_master_reclength,
-                    table->write_set, WRITE_ROWS_EVENT);
+  error= unpack_row(rli, table, m_width, row_start, &m_cols, row_end,
+                    &m_master_reclength, table->write_set, WRITE_ROWS_EVENT);
   bitmap_copy(table->read_set, table->write_set);
   return error;
 }
@@ -6495,7 +6524,7 @@ copy_extra_record_fields(TABLE *table,
                          my_ptrdiff_t master_fields)
 {
   DBUG_PRINT("info", ("Copying to %p "
-                      "from field %d at offset %u "
+                      "from field %ld at offset %u "
                       "to field %d at offset %u",
                       table->record[0],
                       master_fields, master_reclength,
@@ -6595,6 +6624,11 @@ replace_record(THD *thd, TABLE *table,
 
   while ((error= table->file->ha_write_row(table->record[0])))
   {
+    if (error == HA_ERR_LOCK_DEADLOCK || error == HA_ERR_LOCK_WAIT_TIMEOUT)
+    {
+      table->file->print_error(error, MYF(0)); /* to check at exec_relay_log_event */
+      DBUG_RETURN(error);
+    }
     if ((keynum= table->file->get_dup_key(error)) < 0)
     {
       /* We failed to retrieve the duplicate key */
@@ -6649,7 +6683,7 @@ replace_record(THD *thd, TABLE *table,
        present on the master from table->record[1], if there are any.
     */
     copy_extra_record_fields(table, master_reclength, master_fields);
-    
+
     /*
        REPLACE is defined as either INSERT or DELETE + INSERT.  If
        possible, we can replace it with an UPDATE, but that will not
@@ -6728,8 +6762,26 @@ static bool record_compare(TABLE *table)
 
 /*
   Find the row given by 'key', if the table has keys, or else use a table scan
-  to find (and fetch) the row.  If the engine allows random access of the
-  records, a combination of position() and rnd_pos() will be used.
+  to find (and fetch) the row.
+
+  If the engine allows random access of the records, a combination of
+  position() and rnd_pos() will be used.
+
+  @param table Pointer to table to search
+  @param key   Pointer to key to use for search, if table has key
+
+  @pre <code>table->record[0]</code> shall contain the row to locate
+  and <code>key</code> shall contain a key to use for searching, if
+  the engine has a key.
+
+  @post If the return value is zero, <code>table->record[1]</code>
+  will contain the fetched row and the internal "cursor" will refer to
+  the row. If the return value is non-zero,
+  <code>table->record[1]</code> is undefined.  In either case,
+  <code>table->record[0]</code> is undefined.
+
+  @return Zero if the row was successfully fetched into
+  <code>table->record[1]</code>, error code otherwise.
  */
 
 static int find_and_fetch_row(TABLE *table, byte *key)
@@ -6749,13 +6801,28 @@ static int find_and_fetch_row(TABLE *table, byte *key)
       row reference using the position() member function (it will be
       stored in table->file->ref) and the use rnd_pos() to position
       the "cursor" (i.e., record[0] in this case) at the correct row.
+
+      TODO: Add a check that the correct record has been fetched by
+      comparing with the original record. Take into account that the
+      record on the master and slave can be of different
+      length. Something along these lines should work:
+
+      ADD>>>  store_record(table,record[1]);
+              int error= table->file->rnd_pos(table->record[0], table->file->ref);
+      ADD>>>  DBUG_ASSERT(memcmp(table->record[1], table->record[0],
+                                 table->s->reclength) == 0);
+
     */
     table->file->position(table->record[0]);
-    DBUG_RETURN(table->file->rnd_pos(table->record[0], table->file->ref));
+    int error= table->file->rnd_pos(table->record[0], table->file->ref);
+    /*
+      rnd_pos() returns the record in table->record[0], so we have to
+      move it to table->record[1].
+     */
+    bmove_align(table->record[1], table->record[0], table->s->reclength);
+    DBUG_RETURN(error);
   }
 
-  DBUG_ASSERT(table->record[1]);
-
   /* We need to retrieve all fields */
   /* TODO: Move this out from this function to main loop */
   table->use_all_columns();
@@ -6765,7 +6832,16 @@ static int find_and_fetch_row(TABLE *table, byte *key)
     int error;
     /* We have a key: search the table using the index */
     if (!table->file->inited && (error= table->file->ha_index_init(0, FALSE)))
-      return error;
+      DBUG_RETURN(error);
+
+  /*
+    Don't print debug messages when running valgrind since they can
+    trigger false warnings.
+   */
+#ifndef HAVE_purify
+    DBUG_DUMP("table->record[0]", table->record[0], table->s->reclength);
+    DBUG_DUMP("table->record[1]", table->record[1], table->s->reclength);
+#endif
 
     /*
       We need to set the null bytes to ensure that the filler bit are
@@ -6785,6 +6861,14 @@ static int find_and_fetch_row(TABLE *table, byte *key)
       DBUG_RETURN(error);
     }
 
+  /*
+    Don't print debug messages when running valgrind since they can
+    trigger false warnings.
+   */
+#ifndef HAVE_purify
+    DBUG_DUMP("table->record[0]", table->record[0], table->s->reclength);
+    DBUG_DUMP("table->record[1]", table->record[1], table->s->reclength);
+#endif
     /*
       Below is a minor "optimization".  If the key (i.e., key number
       0) has the HA_NOSAME flag set, we know that we have found the
@@ -6969,8 +7053,8 @@ int Delete_rows_log_event::do_after_row_operations(TABLE *table, int error)
 
 int Delete_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
                                           TABLE *table,
-                                          char const *row_start,
-                                          char const **row_end)
+                                          char const *const row_start,
+                                          char const **const row_end)
 {
   int error;
   DBUG_ASSERT(row_start && row_end);
@@ -6980,10 +7064,8 @@ int Delete_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
   */
   DBUG_ASSERT(table->s->fields >= m_width);
 
-  error= unpack_row(rli,
-                    table, m_width, table->record[0], 
-                    row_start, &m_cols, row_end, &m_master_reclength,
-                    table->read_set, DELETE_ROWS_EVENT);
+  error= unpack_row(rli, table, m_width, row_start, &m_cols, row_end,
+                    &m_master_reclength, table->read_set, DELETE_ROWS_EVENT);
   /*
     If we will access rows using the random access method, m_key will
     be set to NULL, so we do not need to make a key copy in that case.
@@ -7106,8 +7188,8 @@ int Update_rows_log_event::do_after_row_operations(TABLE *table, int error)
 
 int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
                                           TABLE *table,
-                                          char const *row_start,
-                                          char const **row_end)
+                                          char const *const row_start,
+                                          char const **const row_end)
 {
   int error;
   DBUG_ASSERT(row_start && row_end);
@@ -7117,21 +7199,31 @@ int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO *rli,
   */
   DBUG_ASSERT(table->s->fields >= m_width);
 
+  /*
+    We need to perform some juggling below since unpack_row() always
+    unpacks into table->record[0]. For more information, see the
+    comments for unpack_row().
+  */
+
   /* record[0] is the before image for the update */
-  error= unpack_row(rli,
-                    table, m_width, table->record[0],
-                    row_start, &m_cols, row_end, &m_master_reclength,
-                    table->read_set, UPDATE_ROWS_EVENT);
-  row_start = *row_end;
+  error= unpack_row(rli, table, m_width, row_start, &m_cols, row_end,
+                    &m_master_reclength, table->read_set, UPDATE_ROWS_EVENT);
+  store_record(table, record[1]);
+  char const *next_start = *row_end;
   /* m_after_image is the after image for the update */
-  error= unpack_row(rli,
-                    table, m_width, m_after_image,
-                    row_start, &m_cols, row_end, &m_master_reclength,
-                    table->write_set, UPDATE_ROWS_EVENT);
+  error= unpack_row(rli, table, m_width, next_start, &m_cols, row_end,
+                    &m_master_reclength, table->write_set, UPDATE_ROWS_EVENT);
+  bmove_align(m_after_image, table->record[0], table->s->reclength);
+  restore_record(table, record[1]);
 
+  /*
+    Don't print debug messages when running valgrind since they can
+    trigger false warnings.
+   */
+#ifndef HAVE_purify
   DBUG_DUMP("record[0]", (const char *)table->record[0], table->s->reclength);
   DBUG_DUMP("m_after_image", (const char *)m_after_image, table->s->reclength);
-
+#endif
 
   /*
     If we will access rows using the random access method, m_key will
diff --git a/sql/log_event.h b/sql/log_event.h
index 81ce2f18b4d..c3f015e723c 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -200,8 +200,26 @@ struct sql_ex_info
 #define EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN (4 + 4 + 4 + 1)
 #define EXECUTE_LOAD_QUERY_HEADER_LEN  (QUERY_HEADER_LEN + EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN)
 
-/*
-   Event header offsets;
+/* 
+  Max number of possible extra bytes in a replication event compared to a
+  packet (i.e. a query) sent from client to master;
+  First, an auxiliary log_event status vars estimation:
+*/
+#define MAX_SIZE_LOG_EVENT_STATUS (4 /* flags2 */   + \
+                                   8 /* sql mode */ + \
+                                   1 + 1 + 255 /* catalog */ + \
+                                   4 /* autoinc */ + \
+                                   6 /* charset */ + \
+                                   MAX_TIME_ZONE_NAME_LENGTH)
+#define MAX_LOG_EVENT_HEADER   ( /* in order of Query_log_event::write */ \
+  LOG_EVENT_HEADER_LEN + /* write_header */ \
+  QUERY_HEADER_LEN     + /* write_data */   \
+  EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN + /*write_post_header_for_derived */ \
+  MAX_SIZE_LOG_EVENT_STATUS + /* status */ \
+  NAME_LEN + 1)
+
+/* 
+   Event header offsets; 
    these point to places inside the fixed header.
 */
 
diff --git a/sql/mysql_priv.h b/sql/mysql_priv.h
index b0947249439..e30da1f2f0b 100644
--- a/sql/mysql_priv.h
+++ b/sql/mysql_priv.h
@@ -105,6 +105,17 @@ extern CHARSET_INFO *system_charset_info, *files_charset_info ;
 extern CHARSET_INFO *national_charset_info, *table_alias_charset;
 
 
+enum Derivation
+{
+  DERIVATION_IGNORABLE= 5,
+  DERIVATION_COERCIBLE= 4,
+  DERIVATION_SYSCONST= 3,
+  DERIVATION_IMPLICIT= 2,
+  DERIVATION_NONE= 1,
+  DERIVATION_EXPLICIT= 0
+};
+
+
 typedef struct my_locale_st
 {
   const char *name;
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 7e57afeafee..16bfbcbc565 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -3160,11 +3160,6 @@ with --log-bin instead.");
   }
   if (global_system_variables.binlog_format == BINLOG_FORMAT_UNSPEC)
   {
-#if defined(HAVE_NDB_BINLOG) && defined(HAVE_ROW_BASED_REPLICATION)
-    if (opt_bin_log && have_ndbcluster == SHOW_OPTION_YES)
-      global_system_variables.binlog_format= BINLOG_FORMAT_ROW;
-    else
-#endif
 #if defined(HAVE_ROW_BASED_REPLICATION)
       global_system_variables.binlog_format= BINLOG_FORMAT_MIXED;
 #else
diff --git a/sql/parse_file.cc b/sql/parse_file.cc
index 0071d59242e..0a2d4012af4 100644
--- a/sql/parse_file.cc
+++ b/sql/parse_file.cc
@@ -368,32 +368,33 @@ my_bool rename_in_schema_file(const char *schema, const char *old_name,
 {
   char old_path[FN_REFLEN], new_path[FN_REFLEN], arc_path[FN_REFLEN];
 
-  strxnmov(old_path, FN_REFLEN-1, mysql_data_home, "/", schema, "/",
-           old_name, reg_ext, NullS);
-  (void) unpack_filename(old_path, old_path);
-
-  strxnmov(new_path, FN_REFLEN-1, mysql_data_home, "/", schema, "/",
-           new_name, reg_ext, NullS);
-  (void) unpack_filename(new_path, new_path);
+  build_table_filename(old_path, sizeof(old_path) - 1,
+                       schema, old_name, reg_ext, 0);
+  build_table_filename(new_path, sizeof(new_path) - 1,
+                       schema, new_name, reg_ext, 0);
 
   if (my_rename(old_path, new_path, MYF(MY_WME)))
     return 1;
 
   /* check if arc_dir exists */
-  strxnmov(arc_path, FN_REFLEN-1, mysql_data_home, "/", schema, "/arc", NullS);
-  (void) unpack_filename(arc_path, arc_path);
+  build_table_filename(arc_path, sizeof(arc_path) - 1, schema, "arc", "", 0);
   
   if (revision > 0 && !access(arc_path, F_OK))
   {
+    char old_name_buf[FN_REFLEN], new_name_buf[FN_REFLEN];
     ulonglong limit= ((revision > num_view_backups) ?
                       revision - num_view_backups : 0);
+
+    VOID(tablename_to_filename(old_name, old_name_buf, sizeof(old_name_buf)));
+    VOID(tablename_to_filename(new_name, new_name_buf, sizeof(new_name_buf)));
+
     for (; revision > limit ; revision--)
     {
       my_snprintf(old_path, FN_REFLEN, "%s/%s%s-%04lu",
-		  arc_path, old_name, reg_ext, (ulong)revision);
+		  arc_path, old_name_buf, reg_ext, (ulong) revision);
       (void) unpack_filename(old_path, old_path);
       my_snprintf(new_path, FN_REFLEN, "%s/%s%s-%04lu",
-		  arc_path, new_name, reg_ext, (ulong)revision);
+		  arc_path, new_name_buf, reg_ext, (ulong) revision);
       (void) unpack_filename(new_path, new_path);
       my_rename(old_path, new_path, MYF(0));
     }
diff --git a/sql/repl_failsafe.cc b/sql/repl_failsafe.cc
index 66e2aa1c31c..2b034d50d6a 100644
--- a/sql/repl_failsafe.cc
+++ b/sql/repl_failsafe.cc
@@ -962,7 +962,7 @@ bool load_master_data(THD* thd)
      Cancel the previous START SLAVE UNTIL, as the fact to download
      a new copy logically makes UNTIL irrelevant.
   */
-  clear_until_condition(&active_mi->rli);
+  active_mi->rli.clear_until_condition();
 
   /*
     No need to update rli.event* coordinates, they will be when the slave
diff --git a/sql/rpl_injector.cc b/sql/rpl_injector.cc
index 5a74fd58755..3a0fca4dfa5 100644
--- a/sql/rpl_injector.cc
+++ b/sql/rpl_injector.cc
@@ -39,6 +39,8 @@ injector::transaction::transaction(MYSQL_BIN_LOG *log, THD *thd)
   m_start_pos.m_file_pos= log_info.pos;
 
   begin_trans(m_thd);
+
+  thd->set_current_stmt_binlog_row_based();
 }
 
 injector::transaction::~transaction()
diff --git a/sql/rpl_mi.cc b/sql/rpl_mi.cc
new file mode 100644
index 00000000000..c89c8aa131e
--- /dev/null
+++ b/sql/rpl_mi.cc
@@ -0,0 +1,386 @@
+/* Copyright (C) 2000-2003 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#include <my_global.h> // For HAVE_REPLICATION
+#include "mysql_priv.h"
+#include <my_dir.h>
+
+#include "rpl_mi.h"
+
+#ifdef HAVE_REPLICATION
+
+
+// Defined in slave.cc
+int init_intvar_from_file(int* var, IO_CACHE* f, int default_val);
+int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
+			  const char *default_val);
+
+MASTER_INFO::MASTER_INFO()
+  :ssl(0), fd(-1),  io_thd(0), inited(0),
+   abort_slave(0),slave_running(0), slave_run_id(0)
+{
+  host[0] = 0; user[0] = 0; password[0] = 0;
+  ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0;
+  ssl_cipher[0]= 0; ssl_key[0]= 0;
+  
+  bzero((char*) &file, sizeof(file));
+  pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
+  pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
+  pthread_cond_init(&data_cond, NULL);
+  pthread_cond_init(&start_cond, NULL);
+  pthread_cond_init(&stop_cond, NULL);
+}
+
+MASTER_INFO::~MASTER_INFO()
+{
+  pthread_mutex_destroy(&run_lock);
+  pthread_mutex_destroy(&data_lock);
+  pthread_cond_destroy(&data_cond);
+  pthread_cond_destroy(&start_cond);
+  pthread_cond_destroy(&stop_cond);
+}
+
+
+void init_master_info_with_options(MASTER_INFO* mi)
+{
+  DBUG_ENTER("init_master_info_with_options");
+
+  mi->master_log_name[0] = 0;
+  mi->master_log_pos = BIN_LOG_HEADER_SIZE;             // skip magic number
+
+  if (master_host)
+    strmake(mi->host, master_host, sizeof(mi->host) - 1);
+  if (master_user)
+    strmake(mi->user, master_user, sizeof(mi->user) - 1);
+  if (master_password)
+    strmake(mi->password, master_password, MAX_PASSWORD_LENGTH);
+  mi->port = master_port;
+  mi->connect_retry = master_connect_retry;
+
+  mi->ssl= master_ssl;
+  if (master_ssl_ca)
+    strmake(mi->ssl_ca, master_ssl_ca, sizeof(mi->ssl_ca)-1);
+  if (master_ssl_capath)
+    strmake(mi->ssl_capath, master_ssl_capath, sizeof(mi->ssl_capath)-1);
+  if (master_ssl_cert)
+    strmake(mi->ssl_cert, master_ssl_cert, sizeof(mi->ssl_cert)-1);
+  if (master_ssl_cipher)
+    strmake(mi->ssl_cipher, master_ssl_cipher, sizeof(mi->ssl_cipher)-1);
+  if (master_ssl_key)
+    strmake(mi->ssl_key, master_ssl_key, sizeof(mi->ssl_key)-1);
+  DBUG_VOID_RETURN;
+}
+
+
+#define LINES_IN_MASTER_INFO_WITH_SSL 14
+
+
+int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
+                     const char* slave_info_fname,
+                     bool abort_if_no_master_info_file,
+                     int thread_mask)
+{
+  int fd,error;
+  char fname[FN_REFLEN+128];
+  DBUG_ENTER("init_master_info");
+
+  if (mi->inited)
+  {
+    /*
+      We have to reset read position of relay-log-bin as we may have
+      already been reading from 'hotlog' when the slave was stopped
+      last time. If this case pos_in_file would be set and we would
+      get a crash when trying to read the signature for the binary
+      relay log.
+
+      We only rewind the read position if we are starting the SQL
+      thread. The handle_slave_sql thread assumes that the read
+      position is at the beginning of the file, and will read the
+      "signature" and then fast-forward to the last position read.
+    */
+    if (thread_mask & SLAVE_SQL)
+    {
+      my_b_seek(mi->rli.cur_log, (my_off_t) 0);
+    }
+    DBUG_RETURN(0);
+  }
+
+  mi->mysql=0;
+  mi->file_id=1;
+  fn_format(fname, master_info_fname, mysql_data_home, "", 4+32);
+
+  /*
+    We need a mutex while we are changing master info parameters to
+    keep other threads from reading bogus info
+  */
+
+  pthread_mutex_lock(&mi->data_lock);
+  fd = mi->fd;
+
+  /* does master.info exist ? */
+
+  if (access(fname,F_OK))
+  {
+    if (abort_if_no_master_info_file)
+    {
+      pthread_mutex_unlock(&mi->data_lock);
+      DBUG_RETURN(0);
+    }
+    /*
+      if someone removed the file from underneath our feet, just close
+      the old descriptor and re-create the old file
+    */
+    if (fd >= 0)
+      my_close(fd, MYF(MY_WME));
+    if ((fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0 )
+    {
+      sql_print_error("Failed to create a new master info file (\
+file '%s', errno %d)", fname, my_errno);
+      goto err;
+    }
+    if (init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,0,
+                      MYF(MY_WME)))
+    {
+      sql_print_error("Failed to create a cache on master info file (\
+file '%s')", fname);
+      goto err;
+    }
+
+    mi->fd = fd;
+    init_master_info_with_options(mi);
+
+  }
+  else // file exists
+  {
+    if (fd >= 0)
+      reinit_io_cache(&mi->file, READ_CACHE, 0L,0,0);
+    else
+    {
+      if ((fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0 )
+      {
+        sql_print_error("Failed to open the existing master info file (\
+file '%s', errno %d)", fname, my_errno);
+        goto err;
+      }
+      if (init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,
+                        0, MYF(MY_WME)))
+      {
+        sql_print_error("Failed to create a cache on master info file (\
+file '%s')", fname);
+        goto err;
+      }
+    }
+
+    mi->fd = fd;
+    int port, connect_retry, master_log_pos, ssl= 0, lines;
+    char *first_non_digit;
+
+    /*
+       Starting from 4.1.x master.info has new format. Now its
+       first line contains number of lines in file. By reading this
+       number we will be always distinguish to which version our
+       master.info corresponds to. We can't simply count lines in
+       file since versions before 4.1.x could generate files with more
+       lines than needed.
+       If first line doesn't contain a number or contain number less than
+       14 then such file is treated like file from pre 4.1.1 version.
+       There is no ambiguity when reading an old master.info, as before
+       4.1.1, the first line contained the binlog's name, which is either
+       empty or has an extension (contains a '.'), so can't be confused
+       with an integer.
+
+       So we're just reading first line and trying to figure which version
+       is this.
+    */
+
+    /*
+       The first row is temporarily stored in mi->master_log_name,
+       if it is line count and not binlog name (new format) it will be
+       overwritten by the second row later.
+    */
+    if (init_strvar_from_file(mi->master_log_name,
+                              sizeof(mi->master_log_name), &mi->file,
+                              ""))
+      goto errwithmsg;
+
+    lines= strtoul(mi->master_log_name, &first_non_digit, 10);
+
+    if (mi->master_log_name[0]!='\0' &&
+        *first_non_digit=='\0' && lines >= LINES_IN_MASTER_INFO_WITH_SSL)
+    {                                          // Seems to be new format
+      if (init_strvar_from_file(mi->master_log_name,
+            sizeof(mi->master_log_name), &mi->file, ""))
+        goto errwithmsg;
+    }
+    else
+      lines= 7;
+
+    if (init_intvar_from_file(&master_log_pos, &mi->file, 4) ||
+        init_strvar_from_file(mi->host, sizeof(mi->host), &mi->file,
+                              master_host) ||
+        init_strvar_from_file(mi->user, sizeof(mi->user), &mi->file,
+                              master_user) ||
+        init_strvar_from_file(mi->password, SCRAMBLED_PASSWORD_CHAR_LENGTH+1,
+                              &mi->file, master_password) ||
+        init_intvar_from_file(&port, &mi->file, master_port) ||
+        init_intvar_from_file(&connect_retry, &mi->file,
+                              master_connect_retry))
+      goto errwithmsg;
+
+    /*
+       If file has ssl part use it even if we have server without
+       SSL support. But these option will be ignored later when
+       slave will try connect to master, so in this case warning
+       is printed.
+     */
+    if (lines >= LINES_IN_MASTER_INFO_WITH_SSL &&
+        (init_intvar_from_file(&ssl, &mi->file, master_ssl) ||
+         init_strvar_from_file(mi->ssl_ca, sizeof(mi->ssl_ca),
+                               &mi->file, master_ssl_ca) ||
+         init_strvar_from_file(mi->ssl_capath, sizeof(mi->ssl_capath),
+                               &mi->file, master_ssl_capath) ||
+         init_strvar_from_file(mi->ssl_cert, sizeof(mi->ssl_cert),
+                               &mi->file, master_ssl_cert) ||
+         init_strvar_from_file(mi->ssl_cipher, sizeof(mi->ssl_cipher),
+                               &mi->file, master_ssl_cipher) ||
+         init_strvar_from_file(mi->ssl_key, sizeof(mi->ssl_key),
+                              &mi->file, master_ssl_key)))
+      goto errwithmsg;
+#ifndef HAVE_OPENSSL
+    if (ssl)
+      sql_print_warning("SSL information in the master info file "
+                      "('%s') are ignored because this MySQL slave was compiled "
+                      "without SSL support.", fname);
+#endif /* HAVE_OPENSSL */
+
+    /*
+      This has to be handled here as init_intvar_from_file can't handle
+      my_off_t types
+    */
+    mi->master_log_pos= (my_off_t) master_log_pos;
+    mi->port= (uint) port;
+    mi->connect_retry= (uint) connect_retry;
+    mi->ssl= (my_bool) ssl;
+  }
+  DBUG_PRINT("master_info",("log_file_name: %s  position: %ld",
+                            mi->master_log_name,
+                            (ulong) mi->master_log_pos));
+
+  mi->rli.mi = mi;
+  if (init_relay_log_info(&mi->rli, slave_info_fname))
+    goto err;
+
+  mi->inited = 1;
+  // now change cache READ -> WRITE - must do this before flush_master_info
+  reinit_io_cache(&mi->file, WRITE_CACHE, 0L, 0, 1);
+  if ((error=test(flush_master_info(mi, 1))))
+    sql_print_error("Failed to flush master info file");
+  pthread_mutex_unlock(&mi->data_lock);
+  DBUG_RETURN(error);
+
+errwithmsg:
+  sql_print_error("Error reading master configuration");
+
+err:
+  if (fd >= 0)
+  {
+    my_close(fd, MYF(0));
+    end_io_cache(&mi->file);
+  }
+  mi->fd= -1;
+  pthread_mutex_unlock(&mi->data_lock);
+  DBUG_RETURN(1);
+}
+
+
+/*
+  RETURN
+     2 - flush relay log failed
+     1 - flush master info failed
+     0 - all ok
+*/
+int flush_master_info(MASTER_INFO* mi, bool flush_relay_log_cache)
+{
+  IO_CACHE* file = &mi->file;
+  char lbuf[22];
+  DBUG_ENTER("flush_master_info");
+  DBUG_PRINT("enter",("master_pos: %ld", (long) mi->master_log_pos));
+
+  /*
+    Flush the relay log to disk. If we don't do it, then the relay log while
+    have some part (its last kilobytes) in memory only, so if the slave server
+    dies now, with, say, from master's position 100 to 150 in memory only (not
+    on disk), and with position 150 in master.info, then when the slave
+    restarts, the I/O thread will fetch binlogs from 150, so in the relay log
+    we will have "[0, 100] U [150, infinity[" and nobody will notice it, so the
+    SQL thread will jump from 100 to 150, and replication will silently break.
+
+    When we come to this place in code, relay log may or not be initialized;
+    the caller is responsible for setting 'flush_relay_log_cache' accordingly.
+  */
+  if (flush_relay_log_cache &&
+      flush_io_cache(mi->rli.relay_log.get_log_file()))
+    DBUG_RETURN(2);
+
+  /*
+    We flushed the relay log BEFORE the master.info file, because if we crash
+    now, we will get a duplicate event in the relay log at restart. If we
+    flushed in the other order, we would get a hole in the relay log.
+    And duplicate is better than hole (with a duplicate, in later versions we
+    can add detection and scrap one event; with a hole there's nothing we can
+    do).
+  */
+
+  /*
+     In certain cases this code may create master.info files that seems
+     corrupted, because of extra lines filled with garbage in the end
+     file (this happens if new contents take less space than previous
+     contents of file). But because of number of lines in the first line
+     of file we don't care about this garbage.
+  */
+
+  my_b_seek(file, 0L);
+  my_b_printf(file, "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n",
+              LINES_IN_MASTER_INFO_WITH_SSL,
+              mi->master_log_name, llstr(mi->master_log_pos, lbuf),
+              mi->host, mi->user,
+              mi->password, mi->port, mi->connect_retry,
+              (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert,
+              mi->ssl_cipher, mi->ssl_key);
+  DBUG_RETURN(-flush_io_cache(file));
+}
+
+
+void end_master_info(MASTER_INFO* mi)
+{
+  DBUG_ENTER("end_master_info");
+
+  if (!mi->inited)
+    DBUG_VOID_RETURN;
+  end_relay_log_info(&mi->rli);
+  if (mi->fd >= 0)
+  {
+    end_io_cache(&mi->file);
+    (void)my_close(mi->fd, MYF(MY_WME));
+    mi->fd = -1;
+  }
+  mi->inited = 0;
+
+  DBUG_VOID_RETURN;
+}
+
+
+#endif /* HAVE_REPLICATION */
diff --git a/sql/rpl_mi.h b/sql/rpl_mi.h
new file mode 100644
index 00000000000..f0a7d6681fe
--- /dev/null
+++ b/sql/rpl_mi.h
@@ -0,0 +1,110 @@
+/* Copyright (C) 2000-2003 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#ifndef RPL_MI_H
+#define RPL_MI_H
+
+#ifdef HAVE_REPLICATION
+
+/*****************************************************************************
+
+  Replication IO Thread
+
+  MASTER_INFO contains:
+    - information about how to connect to a master
+    - current master log name
+    - current master log offset
+    - misc control variables
+
+  MASTER_INFO is initialized once from the master.info file if such
+  exists. Otherwise, data members corresponding to master.info fields
+  are initialized with defaults specified by master-* options. The
+  initialization is done through init_master_info() call.
+
+  The format of master.info file:
+
+  log_name
+  log_pos
+  master_host
+  master_user
+  master_pass
+  master_port
+  master_connect_retry
+
+  To write out the contents of master.info file to disk ( needed every
+  time we read and queue data from the master ), a call to
+  flush_master_info() is required.
+
+  To clean up, call end_master_info()
+
+*****************************************************************************/
+
+class MASTER_INFO
+{
+ public:
+  MASTER_INFO();
+  ~MASTER_INFO();
+
+  /* the variables below are needed because we can change masters on the fly */
+  char master_log_name[FN_REFLEN];
+  char host[HOSTNAME_LENGTH+1];
+  char user[USERNAME_LENGTH+1];
+  char password[MAX_PASSWORD_LENGTH+1];
+  my_bool ssl; // enables use of SSL connection if true
+  char ssl_ca[FN_REFLEN], ssl_capath[FN_REFLEN], ssl_cert[FN_REFLEN];
+  char ssl_cipher[FN_REFLEN], ssl_key[FN_REFLEN];
+
+  my_off_t master_log_pos;
+  File fd; // we keep the file open, so we need to remember the file pointer
+  IO_CACHE file;
+
+  pthread_mutex_t data_lock,run_lock;
+  pthread_cond_t data_cond,start_cond,stop_cond;
+  THD *io_thd;
+  MYSQL* mysql;
+  uint32 file_id;				/* for 3.23 load data infile */
+  RELAY_LOG_INFO rli;
+  uint port;
+  uint connect_retry;
+#ifndef DBUG_OFF
+  int events_till_disconnect;
+#endif
+  bool inited;
+  volatile bool abort_slave;
+  volatile uint slave_running;
+  volatile ulong slave_run_id;
+  /*
+     The difference in seconds between the clock of the master and the clock of
+     the slave (second - first). It must be signed as it may be <0 or >0.
+     clock_diff_with_master is computed when the I/O thread starts; for this the
+     I/O thread does a SELECT UNIX_TIMESTAMP() on the master.
+     "how late the slave is compared to the master" is computed like this:
+     clock_of_slave - last_timestamp_executed_by_SQL_thread - clock_diff_with_master
+
+  */
+  long clock_diff_with_master;
+};
+
+void init_master_info_with_options(MASTER_INFO* mi);
+int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
+		     const char* slave_info_fname,
+		     bool abort_if_no_master_info_file,
+		     int thread_mask);
+void end_master_info(MASTER_INFO* mi);
+int flush_master_info(MASTER_INFO* mi, bool flush_relay_log_cache);
+
+#endif /* HAVE_REPLICATION */
+#endif /* RPL_MI_H */
diff --git a/sql/rpl_rli.cc b/sql/rpl_rli.cc
new file mode 100644
index 00000000000..f01fc5d1c9e
--- /dev/null
+++ b/sql/rpl_rli.cc
@@ -0,0 +1,1112 @@
+/* Copyright (C) 2000-2003 MySQL AB
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 2 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, write to the Free Software
+   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */
+
+#include "mysql_priv.h"
+
+#include "rpl_rli.h"
+#include <my_dir.h>    // For MY_STAT
+#include "sql_repl.h"  // For check_binlog_magic
+
+static int count_relay_log_space(RELAY_LOG_INFO* rli);
+
+// Defined in slave.cc
+int init_intvar_from_file(int* var, IO_CACHE* f, int default_val);
+int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
+			  const char *default_val);
+
+
+st_relay_log_info::st_relay_log_info()
+  :no_storage(FALSE), 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), last_master_timestamp(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), until_condition(UNTIL_NONE),
+   until_log_pos(0), retried_trans(0),
+   tables_to_lock(0), tables_to_lock_count(0),
+   unsafe_to_stop_at(0)
+{
+  DBUG_ENTER("st_relay_log_info::st_relay_log_info");
+
+  group_relay_log_name[0]= event_relay_log_name[0]=
+    group_master_log_name[0]= 0;
+  last_slave_error[0]= until_log_name[0]= ign_master_log_name_end[0]= 0;
+  bzero((char*) &info_file, sizeof(info_file));
+  bzero((char*) &cache_buf, sizeof(cache_buf));
+  cached_charset_invalidate();
+  pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
+  pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
+  pthread_mutex_init(&log_space_lock, MY_MUTEX_INIT_FAST);
+  pthread_cond_init(&data_cond, NULL);
+  pthread_cond_init(&start_cond, NULL);
+  pthread_cond_init(&stop_cond, NULL);
+  pthread_cond_init(&log_space_cond, NULL);
+  relay_log.init_pthread_objects();
+  DBUG_VOID_RETURN;
+}
+
+
+st_relay_log_info::~st_relay_log_info()
+{
+  DBUG_ENTER("st_relay_log_info::~st_relay_log_info");
+
+  pthread_mutex_destroy(&run_lock);
+  pthread_mutex_destroy(&data_lock);
+  pthread_mutex_destroy(&log_space_lock);
+  pthread_cond_destroy(&data_cond);
+  pthread_cond_destroy(&start_cond);
+  pthread_cond_destroy(&stop_cond);
+  pthread_cond_destroy(&log_space_cond);
+  relay_log.cleanup();
+  DBUG_VOID_RETURN;
+}
+
+
+int init_relay_log_info(RELAY_LOG_INFO* rli,
+			const char* info_fname)
+{
+  char fname[FN_REFLEN+128];
+  int info_fd;
+  const char* msg = 0;
+  int error = 0;
+  DBUG_ENTER("init_relay_log_info");
+  DBUG_ASSERT(!rli->no_storage);         // Don't init if there is no storage
+
+  if (rli->inited)                       // Set if this function called
+    DBUG_RETURN(0);
+  fn_format(fname, info_fname, mysql_data_home, "", 4+32);
+  pthread_mutex_lock(&rli->data_lock);
+  info_fd = rli->info_fd;
+  rli->cur_log_fd = -1;
+  rli->slave_skip_counter=0;
+  rli->abort_pos_wait=0;
+  rli->log_space_limit= relay_log_space_limit;
+  rli->log_space_total= 0;
+  rli->tables_to_lock= 0;
+  rli->tables_to_lock_count= 0;
+
+  /*
+    The relay log will now be opened, as a SEQ_READ_APPEND IO_CACHE.
+    Note that the I/O thread flushes it to disk after writing every
+    event, in flush_master_info(mi, 1).
+  */
+
+  /*
+    For the maximum log size, we choose max_relay_log_size if it is
+    non-zero, max_binlog_size otherwise. If later the user does SET
+    GLOBAL on one of these variables, fix_max_binlog_size and
+    fix_max_relay_log_size will reconsider the choice (for example
+    if the user changes max_relay_log_size to zero, we have to
+    switch to using max_binlog_size for the relay log) and update
+    rli->relay_log.max_size (and mysql_bin_log.max_size).
+  */
+  {
+    char buf[FN_REFLEN];
+    const char *ln;
+    static bool name_warning_sent= 0;
+    ln= rli->relay_log.generate_name(opt_relay_logname, "-relay-bin",
+                                     1, buf);
+    /* We send the warning only at startup, not after every RESET SLAVE */
+    if (!opt_relay_logname && !opt_relaylog_index_name && !name_warning_sent)
+    {
+      /*
+        User didn't give us info to name the relay log index file.
+        Picking `hostname`-relay-bin.index like we do, causes replication to
+        fail if this slave's hostname is changed later. So, we would like to
+        instead require a name. But as we don't want to break many existing
+        setups, we only give warning, not error.
+      */
+      sql_print_warning("Neither --relay-log nor --relay-log-index were used;"
+                        " so replication "
+                        "may break when this MySQL server acts as a "
+                        "slave and has his hostname changed!! Please "
+                        "use '--relay-log=%s' to avoid this problem.", ln);
+      name_warning_sent= 1;
+    }
+    /*
+      note, that if open() fails, we'll still have index file open
+      but a destructor will take care of that
+    */
+    if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln) ||
+        rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND, 0,
+                            (max_relay_log_size ? max_relay_log_size :
+                            max_binlog_size), 1))
+    {
+      pthread_mutex_unlock(&rli->data_lock);
+      sql_print_error("Failed in open_log() called from init_relay_log_info()");
+      DBUG_RETURN(1);
+    }
+  }
+
+  /* if file does not exist */
+  if (access(fname,F_OK))
+  {
+    /*
+      If someone removed the file from underneath our feet, just close
+      the old descriptor and re-create the old file
+    */
+    if (info_fd >= 0)
+      my_close(info_fd, MYF(MY_WME));
+    if ((info_fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0)
+    {
+      sql_print_error("Failed to create a new relay log info file (\
+file '%s', errno %d)", fname, my_errno);
+      msg= current_thd->net.last_error;
+      goto err;
+    }
+    if (init_io_cache(&rli->info_file, info_fd, IO_SIZE*2, READ_CACHE, 0L,0,
+                      MYF(MY_WME)))
+    {
+      sql_print_error("Failed to create a cache on relay log info file '%s'",
+                      fname);
+      msg= current_thd->net.last_error;
+      goto err;
+    }
+
+    /* Init relay log with first entry in the relay index file */
+    if (init_relay_log_pos(rli,NullS,BIN_LOG_HEADER_SIZE,0 /* no data lock */,
+                           &msg, 0))
+    {
+      sql_print_error("Failed to open the relay log 'FIRST' (relay_log_pos 4)");
+      goto err;
+    }
+    rli->group_master_log_name[0]= 0;
+    rli->group_master_log_pos= 0;
+    rli->info_fd= info_fd;
+  }
+  else // file exists
+  {
+    if (info_fd >= 0)
+      reinit_io_cache(&rli->info_file, READ_CACHE, 0L,0,0);
+    else
+    {
+      int error=0;
+      if ((info_fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0)
+      {
+        sql_print_error("\
+Failed to open the existing relay log info file '%s' (errno %d)",
+                        fname, my_errno);
+        error= 1;
+      }
+      else if (init_io_cache(&rli->info_file, info_fd,
+                             IO_SIZE*2, READ_CACHE, 0L, 0, MYF(MY_WME)))
+      {
+        sql_print_error("Failed to create a cache on relay log info file '%s'",
+                        fname);
+        error= 1;
+      }
+      if (error)
+      {
+        if (info_fd >= 0)
+          my_close(info_fd, MYF(0));
+        rli->info_fd= -1;
+        rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
+        pthread_mutex_unlock(&rli->data_lock);
+        DBUG_RETURN(1);
+      }
+    }
+
+    rli->info_fd = info_fd;
+    int relay_log_pos, master_log_pos;
+    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->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;
+    }
+    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->group_relay_log_name,
+                           rli->group_relay_log_pos,
+                           0 /* no data lock*/,
+                           &msg, 0))
+    {
+      char llbuf[22];
+      sql_print_error("Failed to open the relay log '%s' (relay_log_pos %s)",
+                      rli->group_relay_log_name,
+                      llstr(rli->group_relay_log_pos, llbuf));
+      goto err;
+    }
+  }
+
+#ifndef DBUG_OFF
+  {
+    char llbuf1[22], llbuf2[22];
+    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
+                        llstr(my_b_tell(rli->cur_log),llbuf1),
+                        llstr(rli->event_relay_log_pos,llbuf2)));
+    DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
+    DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
+  }
+#endif
+
+  /*
+    Now change the cache from READ to WRITE - must do this
+    before flush_relay_log_info
+  */
+  reinit_io_cache(&rli->info_file, WRITE_CACHE,0L,0,1);
+  if ((error= flush_relay_log_info(rli)))
+    sql_print_error("Failed to flush relay log info file");
+  if (count_relay_log_space(rli))
+  {
+    msg="Error counting relay log space";
+    goto err;
+  }
+  rli->inited= 1;
+  pthread_mutex_unlock(&rli->data_lock);
+  DBUG_RETURN(error);
+
+err:
+  sql_print_error(msg);
+  end_io_cache(&rli->info_file);
+  if (info_fd >= 0)
+    my_close(info_fd, MYF(0));
+  rli->info_fd= -1;
+  rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
+  pthread_mutex_unlock(&rli->data_lock);
+  DBUG_RETURN(1);
+}
+
+
+static inline int add_relay_log(RELAY_LOG_INFO* rli,LOG_INFO* linfo)
+{
+  MY_STAT s;
+  DBUG_ENTER("add_relay_log");
+  if (!my_stat(linfo->log_file_name,&s,MYF(0)))
+  {
+    sql_print_error("log %s listed in the index, but failed to stat",
+                    linfo->log_file_name);
+    DBUG_RETURN(1);
+  }
+  rli->log_space_total += s.st_size;
+#ifndef DBUG_OFF
+  char buf[22];
+  DBUG_PRINT("info",("log_space_total: %s", llstr(rli->log_space_total,buf)));
+#endif
+  DBUG_RETURN(0);
+}
+
+
+static int count_relay_log_space(RELAY_LOG_INFO* rli)
+{
+  LOG_INFO linfo;
+  DBUG_ENTER("count_relay_log_space");
+  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");
+    DBUG_RETURN(1);
+  }
+  do
+  {
+    if (add_relay_log(rli,&linfo))
+      DBUG_RETURN(1);
+  } while (!rli->relay_log.find_next_log(&linfo, 1));
+  /*
+     As we have counted everything, including what may have written in a
+     preceding write, we must reset bytes_written, or we may count some space
+     twice.
+  */
+  rli->relay_log.reset_bytes_written();
+  DBUG_RETURN(0);
+}
+
+
+void st_relay_log_info::clear_slave_error()
+{
+  DBUG_ENTER("clear_slave_error");
+
+  /* Clear the errors displayed by SHOW SLAVE STATUS */
+  last_slave_error[0]= 0;
+  last_slave_errno= 0;
+  DBUG_VOID_RETURN;
+}
+
+/*
+   Reset UNTIL condition for RELAY_LOG_INFO
+
+   SYNOPSYS
+    clear_until_condition()
+      rli - RELAY_LOG_INFO structure where UNTIL condition should be reset
+ */
+
+void st_relay_log_info::clear_until_condition()
+{
+  DBUG_ENTER("clear_until_condition");
+
+  until_condition= RELAY_LOG_INFO::UNTIL_NONE;
+  until_log_name[0]= 0;
+  until_log_pos= 0;
+  DBUG_VOID_RETURN;
+}
+
+
+/*
+  Open the given relay log
+
+  SYNOPSIS
+    init_relay_log_pos()
+    rli                 Relay information (will be initialized)
+    log                 Name of relay log file to read from. NULL = First log
+    pos                 Position in relay log file
+    need_data_lock      Set to 1 if this functions should do mutex locks
+    errmsg              Store pointer to error message here
+    look_for_description_event
+                        1 if we should look for such an event. We only need
+                        this when the SQL thread starts and opens an existing
+                        relay log and has to execute it (possibly from an
+                        offset >4); then we need to read the first event of
+                        the relay log to be able to parse the events we have
+                        to execute.
+
+  DESCRIPTION
+  - Close old open relay log files.
+  - If we are using the same relay log as the running IO-thread, then set
+    rli->cur_log to point to the same IO_CACHE entry.
+  - If not, open the 'log' binary file.
+
+  TODO
+    - check proper initialization of group_master_log_name/group_master_log_pos
+
+  RETURN VALUES
+    0   ok
+    1   error.  errmsg is set to point to the error message
+*/
+
+int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log,
+                       ulonglong pos, bool need_data_lock,
+                       const char** errmsg,
+                       bool look_for_description_event)
+{
+  DBUG_ENTER("init_relay_log_pos");
+  DBUG_PRINT("info", ("pos=%lu", pos));
+
+  *errmsg=0;
+  pthread_mutex_t *log_lock=rli->relay_log.get_log_lock();
+
+  if (need_data_lock)
+    pthread_mutex_lock(&rli->data_lock);
+
+  /*
+    Slave threads are not the only users of init_relay_log_pos(). CHANGE MASTER
+    is, too, and init_slave() too; these 2 functions allocate a description
+    event in init_relay_log_pos, which is not freed by the terminating SQL slave
+    thread as that thread is not started by these functions. So we have to free
+    the description_event here, in case, so that there is no memory leak in
+    running, say, CHANGE MASTER.
+  */
+  delete rli->relay_log.description_event_for_exec;
+  /*
+    By default the relay log is in binlog format 3 (4.0).
+    Even if format is 4, this will work enough to read the first event
+    (Format_desc) (remember that format 4 is just lenghtened compared to format
+    3; format 3 is a prefix of format 4).
+  */
+  rli->relay_log.description_event_for_exec= new
+    Format_description_log_event(3);
+
+  pthread_mutex_lock(log_lock);
+
+  /* Close log file and free buffers if it's already open */
+  if (rli->cur_log_fd >= 0)
+  {
+    end_io_cache(&rli->cache_buf);
+    my_close(rli->cur_log_fd, MYF(MY_WME));
+    rli->cur_log_fd = -1;
+  }
+
+  rli->group_relay_log_pos = rli->event_relay_log_pos = pos;
+
+  /*
+    Test to see if the previous run was with the skip of purging
+    If yes, we do not purge when we restart
+  */
+  if (rli->relay_log.find_log_pos(&rli->linfo, NullS, 1))
+  {
+    *errmsg="Could not find first log during relay log initialization";
+    goto err;
+  }
+
+  if (log && rli->relay_log.find_log_pos(&rli->linfo, log, 1))
+  {
+    *errmsg="Could not find target log during relay log initialization";
+    goto err;
+  }
+  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))
+  {
+    /*
+      The IO thread is using this log file.
+      In this case, we will use the same IO_CACHE pointer to
+      read data as the IO thread is using to write data.
+    */
+    my_b_seek((rli->cur_log=rli->relay_log.get_log_file()), (off_t)0);
+    if (check_binlog_magic(rli->cur_log,errmsg))
+      goto err;
+    rli->cur_log_old_open_count=rli->relay_log.get_open_count();
+  }
+  else
+  {
+    /*
+      Open the relay log and set rli->cur_log to point at this one
+    */
+    if ((rli->cur_log_fd=open_binlog(&rli->cache_buf,
+                                     rli->linfo.log_file_name,errmsg)) < 0)
+      goto err;
+    rli->cur_log = &rli->cache_buf;
+  }
+  /*
+    In all cases, check_binlog_magic() has been called so we're at offset 4 for
+    sure.
+  */
+  if (pos > BIN_LOG_HEADER_SIZE) /* If pos<=4, we stay at 4 */
+  {
+    Log_event* ev;
+    while (look_for_description_event)
+    {
+      /*
+        Read the possible Format_description_log_event; if position
+        was 4, no need, it will be read naturally.
+      */
+      DBUG_PRINT("info",("looking for a Format_description_log_event"));
+
+      if (my_b_tell(rli->cur_log) >= pos)
+        break;
+
+      /*
+        Because of we have rli->data_lock and log_lock, we can safely read an
+        event
+      */
+      if (!(ev=Log_event::read_log_event(rli->cur_log,0,
+                                         rli->relay_log.description_event_for_exec)))
+      {
+        DBUG_PRINT("info",("could not read event, rli->cur_log->error=%d",
+                           rli->cur_log->error));
+        if (rli->cur_log->error) /* not EOF */
+        {
+          *errmsg= "I/O error reading event at position 4";
+          goto err;
+        }
+        break;
+      }
+      else if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
+      {
+        DBUG_PRINT("info",("found Format_description_log_event"));
+        delete rli->relay_log.description_event_for_exec;
+        rli->relay_log.description_event_for_exec= (Format_description_log_event*) ev;
+        /*
+          As ev was returned by read_log_event, it has passed is_valid(), so
+          my_malloc() in ctor worked, no need to check again.
+        */
+        /*
+          Ok, we found a Format_description event. But it is not sure that this
+          describes the whole relay log; indeed, one can have this sequence
+          (starting from position 4):
+          Format_desc (of slave)
+          Rotate (of master)
+          Format_desc (of master)
+          So the Format_desc which really describes the rest of the relay log
+          is the 3rd event (it can't be further than that, because we rotate
+          the relay log when we queue a Rotate event from the master).
+          But what describes the Rotate is the first Format_desc.
+          So what we do is:
+          go on searching for Format_description events, until you exceed the
+          position (argument 'pos') or until you find another event than Rotate
+          or Format_desc.
+        */
+      }
+      else
+      {
+        DBUG_PRINT("info",("found event of another type=%d",
+                           ev->get_type_code()));
+        look_for_description_event= (ev->get_type_code() == ROTATE_EVENT);
+        delete ev;
+      }
+    }
+    my_b_seek(rli->cur_log,(off_t)pos);
+#ifndef DBUG_OFF
+  {
+    char llbuf1[22], llbuf2[22];
+    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
+                        llstr(my_b_tell(rli->cur_log),llbuf1),
+                        llstr(rli->event_relay_log_pos,llbuf2)));
+  }
+#endif
+
+  }
+
+err:
+  /*
+    If we don't purge, we can't honour relay_log_space_limit ;
+    silently discard it
+  */
+  if (!relay_log_purge)
+    rli->log_space_limit= 0;
+  pthread_cond_broadcast(&rli->data_cond);
+
+  pthread_mutex_unlock(log_lock);
+
+  if (need_data_lock)
+    pthread_mutex_unlock(&rli->data_lock);
+  if (!rli->relay_log.description_event_for_exec->is_valid() && !*errmsg)
+    *errmsg= "Invalid Format_description log event; could be out of memory";
+
+  DBUG_RETURN ((*errmsg) ? 1 : 0);
+}
+
+
+/*
+  Waits until the SQL thread reaches (has executed up to) the
+  log/position or timed out.
+
+  SYNOPSIS
+    wait_for_pos()
+    thd             client thread that sent SELECT MASTER_POS_WAIT
+    log_name        log name to wait for
+    log_pos         position to wait for
+    timeout         timeout in seconds before giving up waiting
+
+  NOTES
+    timeout is longlong whereas it should be ulong ; but this is
+    to catch if the user submitted a negative timeout.
+
+  RETURN VALUES
+    -2          improper arguments (log_pos<0)
+                or slave not running, or master info changed
+                during the function's execution,
+                or client thread killed. -2 is translated to NULL by caller
+    -1          timed out
+    >=0         number of log events the function had to wait
+                before reaching the desired log/position
+ */
+
+int st_relay_log_info::wait_for_pos(THD* thd, String* log_name,
+                                    longlong log_pos,
+                                    longlong timeout)
+{
+  int event_count = 0;
+  ulong init_abort_pos_wait;
+  int error=0;
+  struct timespec abstime; // for timeout checking
+  const char *msg;
+  DBUG_ENTER("st_relay_log_info::wait_for_pos");
+
+  if (!inited)
+    DBUG_RETURN(-1);
+
+  DBUG_PRINT("enter",("log_name: '%s'  log_pos: %lu  timeout: %lu",
+                      log_name->c_ptr(), (ulong) log_pos, (ulong) timeout));
+
+  set_timespec(abstime,timeout);
+  pthread_mutex_lock(&data_lock);
+  msg= thd->enter_cond(&data_cond, &data_lock,
+                       "Waiting for the slave SQL thread to "
+                       "advance position");
+  /*
+     This function will abort when it notices that some CHANGE MASTER or
+     RESET MASTER has changed the master info.
+     To catch this, these commands modify abort_pos_wait ; We just monitor
+     abort_pos_wait and see if it has changed.
+     Why do we have this mechanism instead of simply monitoring slave_running
+     in the loop (we do this too), as CHANGE MASTER/RESET SLAVE require that
+     the SQL thread be stopped?
+     This is becasue if someones does:
+     STOP SLAVE;CHANGE MASTER/RESET SLAVE; START SLAVE;
+     the change may happen very quickly and we may not notice that
+     slave_running briefly switches between 1/0/1.
+  */
+  init_abort_pos_wait= abort_pos_wait;
+
+  /*
+    We'll need to
+    handle all possible log names comparisons (e.g. 999 vs 1000).
+    We use ulong for string->number conversion ; this is no
+    stronger limitation than in find_uniq_filename in sql/log.cc
+  */
+  ulong log_name_extension;
+  char log_name_tmp[FN_REFLEN]; //make a char[] from String
+
+  strmake(log_name_tmp, log_name->ptr(), min(log_name->length(), FN_REFLEN-1));
+
+  char *p= fn_ext(log_name_tmp);
+  char *p_end;
+  if (!*p || log_pos<0)
+  {
+    error= -2; //means improper arguments
+    goto err;
+  }
+  // Convert 0-3 to 4
+  log_pos= max(log_pos, BIN_LOG_HEADER_SIZE);
+  /* p points to '.' */
+  log_name_extension= strtoul(++p, &p_end, 10);
+  /*
+    p_end points to the first invalid character.
+    If it equals to p, no digits were found, error.
+    If it contains '\0' it means conversion went ok.
+  */
+  if (p_end==p || *p_end)
+  {
+    error= -2;
+    goto err;
+  }
+
+  /* The "compare and wait" main loop */
+  while (!thd->killed &&
+         init_abort_pos_wait == abort_pos_wait &&
+         slave_running)
+  {
+    bool pos_reached;
+    int cmp_result= 0;
+
+    DBUG_PRINT("info",
+               ("init_abort_pos_wait: %ld  abort_pos_wait: %ld",
+                init_abort_pos_wait, abort_pos_wait));
+    DBUG_PRINT("info",("group_master_log_name: '%s'  pos: %lu",
+                       group_master_log_name, (ulong) group_master_log_pos));
+
+    /*
+      group_master_log_name can be "", if we are just after a fresh
+      replication start or after a CHANGE MASTER TO MASTER_HOST/PORT
+      (before we have executed one Rotate event from the master) or
+      (rare) if the user is doing a weird slave setup (see next
+      paragraph).  If group_master_log_name is "", we assume we don't
+      have enough info to do the comparison yet, so we just wait until
+      more data. In this case master_log_pos is always 0 except if
+      somebody (wrongly) sets this slave to be a slave of itself
+      without using --replicate-same-server-id (an unsupported
+      configuration which does nothing), then group_master_log_pos
+      will grow and group_master_log_name will stay "".
+    */
+    if (*group_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,
+        and protect against user's input error :
+        if the names do not match up to '.' included, return error
+      */
+      char *q= (char*)(fn_ext(basename)+1);
+      if (strncmp(basename, log_name_tmp, (int)(q-basename)))
+      {
+        error= -2;
+        break;
+      }
+      // Now compare extensions.
+      char *q_end;
+      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= (group_master_log_name_extension > log_name_extension) ? 1 : 0 ;
+
+      pos_reached= ((!cmp_result && group_master_log_pos >= (ulonglong)log_pos) ||
+                    cmp_result > 0);
+      if (pos_reached || thd->killed)
+        break;
+    }
+
+    //wait for master update, with optional timeout.
+
+    DBUG_PRINT("info",("Waiting for master update"));
+    /*
+      We are going to pthread_cond_(timed)wait(); if the SQL thread stops it
+      will wake us up.
+    */
+    if (timeout > 0)
+    {
+      /*
+        Note that pthread_cond_timedwait checks for the timeout
+        before for the condition ; i.e. it returns ETIMEDOUT
+        if the system time equals or exceeds the time specified by abstime
+        before the condition variable is signaled or broadcast, _or_ if
+        the absolute time specified by abstime has already passed at the time
+        of the call.
+        For that reason, pthread_cond_timedwait will do the "timeoutting" job
+        even if its condition is always immediately signaled (case of a loaded
+        master).
+      */
+      error=pthread_cond_timedwait(&data_cond, &data_lock, &abstime);
+    }
+    else
+      pthread_cond_wait(&data_cond, &data_lock);
+    DBUG_PRINT("info",("Got signal of master update or timed out"));
+    if (error == ETIMEDOUT || error == ETIME)
+    {
+      error= -1;
+      break;
+    }
+    error=0;
+    event_count++;
+    DBUG_PRINT("info",("Testing if killed or SQL thread not running"));
+  }
+
+err:
+  thd->exit_cond(msg);
+  DBUG_PRINT("exit",("killed: %d  abort: %d  slave_running: %d \
+improper_arguments: %d  timed_out: %d",
+                     thd->killed_errno(),
+                     (int) (init_abort_pos_wait != abort_pos_wait),
+                     (int) slave_running,
+                     (int) (error == -2),
+                     (int) (error == -1)));
+  if (thd->killed || init_abort_pos_wait != abort_pos_wait ||
+      !slave_running)
+  {
+    error= -2;
+  }
+  DBUG_RETURN( error ? error : event_count );
+}
+
+
+void st_relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
+                                                bool skip_lock)
+{
+  DBUG_ENTER("st_relay_log_info::inc_group_relay_log_pos");
+
+  if (!skip_lock)
+    pthread_mutex_lock(&data_lock);
+  inc_event_relay_log_pos();
+  group_relay_log_pos= event_relay_log_pos;
+  strmake(group_relay_log_name,event_relay_log_name,
+          sizeof(group_relay_log_name)-1);
+
+  notify_group_relay_log_name_update();
+
+  /*
+    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).
+  */
+  /*
+    In 4.x we used the event's len to compute the positions here. This is
+    wrong if the event was 3.23/4.0 and has been converted to 5.0, because
+    then the event's len is not what is was in the master's binlog, so this
+    will make a wrong group_master_log_pos (yes it's a bug in 3.23->4.0
+    replication: Exec_master_log_pos is wrong). Only way to solve this is to
+    have the original offset of the end of the event the relay log. This is
+    what we do in 5.0: log_pos has become "end_log_pos" (because the real use
+    of log_pos in 4.0 was to compute the end_log_pos; so better to store
+    end_log_pos instead of begin_log_pos.
+    If we had not done this fix here, the problem would also have appeared
+    when the slave and master are 5.0 but with different event length (for
+    example the slave is more recent than the master and features the event
+    UID). It would give false MASTER_POS_WAIT, false Exec_master_log_pos in
+    SHOW SLAVE STATUS, and so the user would do some CHANGE MASTER using this
+    value which would lead to badly broken replication.
+    Even the relay_log_pos will be corrupted in this case, because the len is
+    the relay log is not "val".
+    With the end_log_pos solution, we avoid computations involving lengthes.
+  */
+  DBUG_PRINT("info", ("log_pos: %lu  group_master_log_pos: %lu",
+                      (long) log_pos, (long) group_master_log_pos));
+  if (log_pos) // 3.23 binlogs don't have log_posx
+  {
+    group_master_log_pos= log_pos;
+  }
+  pthread_cond_broadcast(&data_cond);
+  if (!skip_lock)
+    pthread_mutex_unlock(&data_lock);
+  DBUG_VOID_RETURN;
+}
+
+
+void st_relay_log_info::close_temporary_tables()
+{
+  TABLE *table,*next;
+  DBUG_ENTER("st_relay_log_info::close_temporary_tables");
+
+  for (table=save_temporary_tables ; table ; table=next)
+  {
+    next=table->next;
+    /*
+      Don't ask for disk deletion. For now, anyway they will be deleted when
+      slave restarts, but it is a better intention to not delete them.
+    */
+    DBUG_PRINT("info", ("table: %p", table));
+    close_temporary(table, 1, 0);
+  }
+  save_temporary_tables= 0;
+  slave_open_temp_tables= 0;
+  DBUG_VOID_RETURN;
+}
+
+/*
+  purge_relay_logs()
+
+  NOTES
+    Assumes to have a run lock on rli and that no slave thread are running.
+*/
+
+int purge_relay_logs(RELAY_LOG_INFO* rli, THD *thd, bool just_reset,
+                     const char** errmsg)
+{
+  int error=0;
+  DBUG_ENTER("purge_relay_logs");
+
+  /*
+    Even if rli->inited==0, we still try to empty rli->master_log_* variables.
+    Indeed, rli->inited==0 does not imply that they already are empty.
+    It could be that slave's info initialization partly succeeded :
+    for example if relay-log.info existed but *relay-bin*.*
+    have been manually removed, init_relay_log_info reads the old
+    relay-log.info and fills rli->master_log_*, then init_relay_log_info
+    checks for the existence of the relay log, this fails and
+    init_relay_log_info leaves rli->inited to 0.
+    In that pathological case, rli->master_log_pos* will be properly reinited
+    at the next START SLAVE (as RESET SLAVE or CHANGE
+    MASTER, the callers of purge_relay_logs, will delete bogus *.info files
+    or replace them with correct files), however if the user does SHOW SLAVE
+    STATUS before START SLAVE, he will see old, confusing rli->master_log_*.
+    In other words, we reinit rli->master_log_* for SHOW SLAVE STATUS
+    to display fine in any case.
+  */
+
+  rli->group_master_log_name[0]= 0;
+  rli->group_master_log_pos= 0;
+
+  if (!rli->inited)
+  {
+    DBUG_PRINT("info", ("rli->inited == 0"));
+    DBUG_RETURN(0);
+  }
+
+  DBUG_ASSERT(rli->slave_running == 0);
+  DBUG_ASSERT(rli->mi->slave_running == 0);
+
+  rli->slave_skip_counter=0;
+  pthread_mutex_lock(&rli->data_lock);
+
+  /*
+    we close the relay log fd possibly left open by the slave SQL thread,
+    to be able to delete it; the relay log fd possibly left open by the slave
+    I/O thread will be closed naturally in reset_logs() by the
+    close(LOG_CLOSE_TO_BE_OPENED) call
+  */
+  if (rli->cur_log_fd >= 0)
+  {
+    end_io_cache(&rli->cache_buf);
+    my_close(rli->cur_log_fd, MYF(MY_WME));
+    rli->cur_log_fd= -1;
+  }
+
+  if (rli->relay_log.reset_logs(thd))
+  {
+    *errmsg = "Failed during log reset";
+    error=1;
+    goto err;
+  }
+  /* Save name of used relay log file */
+  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);
+  rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE;
+  if (count_relay_log_space(rli))
+  {
+    *errmsg= "Error counting relay log space";
+    goto err;
+  }
+  if (!just_reset)
+    error= init_relay_log_pos(rli, rli->group_relay_log_name,
+                              rli->group_relay_log_pos,
+                              0 /* do not need data lock */, errmsg, 0);
+
+err:
+#ifndef DBUG_OFF
+  char buf[22];
+#endif
+  DBUG_PRINT("info",("log_space_total: %s",llstr(rli->log_space_total,buf)));
+  pthread_mutex_unlock(&rli->data_lock);
+  DBUG_RETURN(error);
+}
+
+
+/*
+     Check if condition stated in UNTIL clause of START SLAVE is reached.
+   SYNOPSYS
+     st_relay_log_info::is_until_satisfied()
+   DESCRIPTION
+     Checks if UNTIL condition is reached. Uses caching result of last
+     comparison of current log file name and target log file name. So cached
+     value should be invalidated if current log file name changes
+     (see st_relay_log_info::notify_... functions).
+
+     This caching is needed to avoid of expensive string comparisons and
+     strtol() conversions needed for log names comparison. We don't need to
+     compare them each time this function is called, we only need to do this
+     when current log name changes. If we have UNTIL_MASTER_POS condition we
+     need to do this only after Rotate_log_event::exec_event() (which is
+     rare, so caching gives real benifit), and if we have UNTIL_RELAY_POS
+     condition then we should invalidate cached comarison value after
+     inc_group_relay_log_pos() which called for each group of events (so we
+     have some benefit if we have something like queries that use
+     autoincrement or if we have transactions).
+
+     Should be called ONLY if until_condition != UNTIL_NONE !
+   RETURN VALUE
+     true - condition met or error happened (condition seems to have
+            bad log file name)
+     false - condition not met
+*/
+
+bool st_relay_log_info::is_until_satisfied()
+{
+  const char *log_name;
+  ulonglong log_pos;
+  DBUG_ENTER("st_relay_log_info::is_until_satisfied");
+
+  DBUG_ASSERT(until_condition != UNTIL_NONE);
+
+  if (until_condition == UNTIL_MASTER_POS)
+  {
+    log_name= group_master_log_name;
+    log_pos= group_master_log_pos;
+  }
+  else
+  { /* until_condition == UNTIL_RELAY_POS */
+    log_name= group_relay_log_name;
+    log_pos= group_relay_log_pos;
+  }
+
+  if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN)
+  {
+    /*
+      We have no cached comparison results so we should compare log names
+      and cache result.
+      If we are after RESET SLAVE, and the SQL slave thread has not processed
+      any event yet, it could be that group_master_log_name is "". In that case,
+      just wait for more events (as there is no sensible comparison to do).
+    */
+
+    if (*log_name)
+    {
+      const char *basename= log_name + dirname_length(log_name);
+
+      const char *q= (const char*)(fn_ext(basename)+1);
+      if (strncmp(basename, until_log_name, (int)(q-basename)) == 0)
+      {
+        /* Now compare extensions. */
+        char *q_end;
+        ulong log_name_extension= strtoul(q, &q_end, 10);
+        if (log_name_extension < until_log_name_extension)
+          until_log_names_cmp_result= UNTIL_LOG_NAMES_CMP_LESS;
+        else
+          until_log_names_cmp_result=
+            (log_name_extension > until_log_name_extension) ?
+            UNTIL_LOG_NAMES_CMP_GREATER : UNTIL_LOG_NAMES_CMP_EQUAL ;
+      }
+      else
+      {
+        /* Probably error so we aborting */
+        sql_print_error("Slave SQL thread is stopped because UNTIL "
+                        "condition is bad.");
+        DBUG_RETURN(TRUE);
+      }
+    }
+    else
+      DBUG_RETURN(until_log_pos == 0);
+  }
+
+  DBUG_RETURN(((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL &&
+           log_pos >= until_log_pos) ||
+          until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER));
+}
+
+
+void st_relay_log_info::cached_charset_invalidate()
+{
+  DBUG_ENTER("st_relay_log_info::cached_charset_invalidate");
+
+  /* Full of zeroes means uninitialized. */
+  bzero(cached_charset, sizeof(cached_charset));
+  DBUG_VOID_RETURN;
+}
+
+
+bool st_relay_log_info::cached_charset_compare(char *charset)
+{
+  DBUG_ENTER("st_relay_log_info::cached_charset_compare");
+
+  if (bcmp(cached_charset, charset, sizeof(cached_charset)))
+  {
+    memcpy(cached_charset, charset, sizeof(cached_charset));
+    DBUG_RETURN(1);
+  }
+  DBUG_RETURN(0);
+}
+
+
+void st_relay_log_info::transaction_end(THD* thd)
+{
+  DBUG_ENTER("st_relay_log_info::transaction_end");
+
+  /*
+    Nothing to do here right now.
+   */
+
+  DBUG_VOID_RETURN;
+}
+
+#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
+void st_relay_log_info::cleanup_context(THD *thd, bool error)
+{
+  DBUG_ENTER("st_relay_log_info::cleanup_context");
+
+  DBUG_ASSERT(sql_thd == thd);
+  /*
+    1) Instances of Table_map_log_event, if ::exec_event() was called on them,
+    may have opened tables, which we cannot be sure have been closed (because
+    maybe the Rows_log_event have not been found or will not be, because slave
+    SQL thread is stopping, or relay log has a missing tail etc). So we close
+    all thread's tables. And so the table mappings have to be cancelled.
+    2) Rows_log_event::exec_event() may even have started statements or
+    transactions on them, which we need to rollback in case of error.
+    3) If finding a Format_description_log_event after a BEGIN, we also need
+    to rollback before continuing with the next events.
+    4) so we need this "context cleanup" function.
+  */
+  if (error)
+  {
+    ha_autocommit_or_rollback(thd, 1); // if a "statement transaction"
+    end_trans(thd, ROLLBACK); // if a "real transaction"
+  }
+  m_table_map.clear_tables();
+  close_thread_tables(thd);
+  clear_tables_to_lock();
+  unsafe_to_stop_at= 0;
+  DBUG_VOID_RETURN;
+}
+#endif
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index 392f12c2a71..d737055baf2 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -21,6 +21,7 @@
 
 #include "rpl_tblmap.h"
 
+
 /****************************************************************************
 
   Replication SQL Thread
@@ -98,8 +99,8 @@ typedef struct st_relay_log_info
   */
   pthread_cond_t start_cond, stop_cond, data_cond;
 
-  /* parent master info structure */
-  struct st_master_info *mi;
+  /* parent MASTER_INFO structure */
+  class MASTER_INFO *mi;
 
   /*
     Needed to deal properly with cur_log getting closed and re-opened with
@@ -164,6 +165,9 @@ typedef struct st_relay_log_info
 
   time_t last_master_timestamp; 
 
+  void clear_slave_error();
+  void clear_until_condition();
+
   /*
     Needed for problems when slave stops and we want to restart it
     skipping one or more events in the master log that have caused
@@ -289,22 +293,6 @@ typedef struct st_relay_log_info
   void cached_charset_invalidate();
   bool cached_charset_compare(char *charset);
 
-  /*
-    To reload special tables when they are changes, we introduce a set
-    of functions that will mark whenever special functions need to be
-    called after modifying tables.  Right now, the tables are either
-    ACL tables or grants tables.
-  */
-  enum enum_reload_flag
-  {
-    RELOAD_NONE_F   = 0UL,
-    RELOAD_GRANT_F  = (1UL << 0), 
-    RELOAD_ACCESS_F = (1UL << 1)
-  };
-
-  ulong m_reload_flags;
-
-  void touching_table(char const* db, char const* table, ulong table_id);
   void transaction_end(THD*);
 
   void cleanup_context(THD *, bool);
@@ -322,4 +310,9 @@ typedef struct st_relay_log_info
   time_t unsafe_to_stop_at;
 } RELAY_LOG_INFO;
 
+
+// Defined in rpl_rli.cc
+int init_relay_log_info(RELAY_LOG_INFO* rli, const char* info_fname);
+
+
 #endif /* RPL_RLI_H */
diff --git a/sql/slave.cc b/sql/slave.cc
index 00d6d168fb8..b497b7c8520 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -31,13 +31,15 @@
 
 #include "rpl_tblmap.h"
 
+int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len);
+
+
 #define MAX_SLAVE_RETRY_PAUSE 5
 bool use_slave_mask = 0;
 MY_BITMAP slave_error_mask;
 
 typedef bool (*CHECK_KILLED_FUNC)(THD*,void*);
 
-volatile bool slave_sql_running = 0, slave_io_running = 0;
 char* slave_load_tmpdir = 0;
 MASTER_INFO *active_mi= 0;
 my_bool replicate_same_server_id;
@@ -59,7 +61,6 @@ static int process_io_create_file(MASTER_INFO* mi, Create_file_log_event* cev);
 static bool wait_for_relay_log_space(RELAY_LOG_INFO* rli);
 static inline bool io_slave_killed(THD* thd,MASTER_INFO* mi);
 static inline bool sql_slave_killed(THD* thd,RELAY_LOG_INFO* rli);
-static int count_relay_log_space(RELAY_LOG_INFO* rli);
 static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type);
 static int safe_connect(THD* thd, MYSQL* mysql, MASTER_INFO* mi);
 static int safe_reconnect(THD* thd, MYSQL* mysql, MASTER_INFO* mi,
@@ -202,223 +203,6 @@ err:
 
 
 /*
-  Open the given relay log
-
-  SYNOPSIS
-    init_relay_log_pos()
-    rli                 Relay information (will be initialized)
-    log                 Name of relay log file to read from. NULL = First log
-    pos                 Position in relay log file
-    need_data_lock      Set to 1 if this functions should do mutex locks
-    errmsg              Store pointer to error message here
-    look_for_description_event
-                        1 if we should look for such an event. We only need
-                        this when the SQL thread starts and opens an existing
-                        relay log and has to execute it (possibly from an
-                        offset >4); then we need to read the first event of
-                        the relay log to be able to parse the events we have
-                        to execute.
-
-  DESCRIPTION
-  - Close old open relay log files.
-  - If we are using the same relay log as the running IO-thread, then set
-    rli->cur_log to point to the same IO_CACHE entry.
-  - If not, open the 'log' binary file.
-
-  TODO
-    - check proper initialization of group_master_log_name/group_master_log_pos
-
-  RETURN VALUES
-    0   ok
-    1   error.  errmsg is set to point to the error message
-*/
-
-int init_relay_log_pos(RELAY_LOG_INFO* rli,const char* log,
-                       ulonglong pos, bool need_data_lock,
-                       const char** errmsg,
-                       bool look_for_description_event)
-{
-  DBUG_ENTER("init_relay_log_pos");
-  DBUG_PRINT("info", ("pos=%lu", pos));
-
-  *errmsg=0;
-  pthread_mutex_t *log_lock=rli->relay_log.get_log_lock();
-
-  if (need_data_lock)
-    pthread_mutex_lock(&rli->data_lock);
-
-  /*
-    Slave threads are not the only users of init_relay_log_pos(). CHANGE MASTER
-    is, too, and init_slave() too; these 2 functions allocate a description
-    event in init_relay_log_pos, which is not freed by the terminating SQL slave
-    thread as that thread is not started by these functions. So we have to free
-    the description_event here, in case, so that there is no memory leak in
-    running, say, CHANGE MASTER.
-  */
-  delete rli->relay_log.description_event_for_exec;
-  /*
-    By default the relay log is in binlog format 3 (4.0).
-    Even if format is 4, this will work enough to read the first event
-    (Format_desc) (remember that format 4 is just lenghtened compared to format
-    3; format 3 is a prefix of format 4).
-  */
-  rli->relay_log.description_event_for_exec= new
-    Format_description_log_event(3);
-
-  pthread_mutex_lock(log_lock);
-
-  /* Close log file and free buffers if it's already open */
-  if (rli->cur_log_fd >= 0)
-  {
-    end_io_cache(&rli->cache_buf);
-    my_close(rli->cur_log_fd, MYF(MY_WME));
-    rli->cur_log_fd = -1;
-  }
-
-  rli->group_relay_log_pos = rli->event_relay_log_pos = pos;
-
-  /*
-    Test to see if the previous run was with the skip of purging
-    If yes, we do not purge when we restart
-  */
-  if (rli->relay_log.find_log_pos(&rli->linfo, NullS, 1))
-  {
-    *errmsg="Could not find first log during relay log initialization";
-    goto err;
-  }
-
-  if (log && rli->relay_log.find_log_pos(&rli->linfo, log, 1))
-  {
-    *errmsg="Could not find target log during relay log initialization";
-    goto err;
-  }
-  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))
-  {
-    /*
-      The IO thread is using this log file.
-      In this case, we will use the same IO_CACHE pointer to
-      read data as the IO thread is using to write data.
-    */
-    my_b_seek((rli->cur_log=rli->relay_log.get_log_file()), (off_t)0);
-    if (check_binlog_magic(rli->cur_log,errmsg))
-      goto err;
-    rli->cur_log_old_open_count=rli->relay_log.get_open_count();
-  }
-  else
-  {
-    /*
-      Open the relay log and set rli->cur_log to point at this one
-    */
-    if ((rli->cur_log_fd=open_binlog(&rli->cache_buf,
-                                     rli->linfo.log_file_name,errmsg)) < 0)
-      goto err;
-    rli->cur_log = &rli->cache_buf;
-  }
-  /*
-    In all cases, check_binlog_magic() has been called so we're at offset 4 for
-    sure.
-  */
-  if (pos > BIN_LOG_HEADER_SIZE) /* If pos<=4, we stay at 4 */
-  {
-    Log_event* ev;
-    while (look_for_description_event)
-    {
-      /*
-        Read the possible Format_description_log_event; if position
-        was 4, no need, it will be read naturally.
-      */
-      DBUG_PRINT("info",("looking for a Format_description_log_event"));
-
-      if (my_b_tell(rli->cur_log) >= pos)
-        break;
-
-      /*
-        Because of we have rli->data_lock and log_lock, we can safely read an
-        event
-      */
-      if (!(ev=Log_event::read_log_event(rli->cur_log,0,
-                                         rli->relay_log.description_event_for_exec)))
-      {
-        DBUG_PRINT("info",("could not read event, rli->cur_log->error=%d",
-                           rli->cur_log->error));
-        if (rli->cur_log->error) /* not EOF */
-        {
-          *errmsg= "I/O error reading event at position 4";
-          goto err;
-        }
-        break;
-      }
-      else if (ev->get_type_code() == FORMAT_DESCRIPTION_EVENT)
-      {
-        DBUG_PRINT("info",("found Format_description_log_event"));
-        delete rli->relay_log.description_event_for_exec;
-        rli->relay_log.description_event_for_exec= (Format_description_log_event*) ev;
-        /*
-          As ev was returned by read_log_event, it has passed is_valid(), so
-          my_malloc() in ctor worked, no need to check again.
-        */
-        /*
-          Ok, we found a Format_description event. But it is not sure that this
-          describes the whole relay log; indeed, one can have this sequence
-          (starting from position 4):
-          Format_desc (of slave)
-          Rotate (of master)
-          Format_desc (of master)
-          So the Format_desc which really describes the rest of the relay log
-          is the 3rd event (it can't be further than that, because we rotate
-          the relay log when we queue a Rotate event from the master).
-          But what describes the Rotate is the first Format_desc.
-          So what we do is:
-          go on searching for Format_description events, until you exceed the
-          position (argument 'pos') or until you find another event than Rotate
-          or Format_desc.
-        */
-      }
-      else
-      {
-        DBUG_PRINT("info",("found event of another type=%d",
-                           ev->get_type_code()));
-        look_for_description_event= (ev->get_type_code() == ROTATE_EVENT);
-        delete ev;
-      }
-    }
-    my_b_seek(rli->cur_log,(off_t)pos);
-#ifndef DBUG_OFF
-  {
-    char llbuf1[22], llbuf2[22];
-    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
-                        llstr(my_b_tell(rli->cur_log),llbuf1),
-                        llstr(rli->event_relay_log_pos,llbuf2)));
-  }
-#endif
-
-  }
-
-err:
-  /*
-    If we don't purge, we can't honour relay_log_space_limit ;
-    silently discard it
-  */
-  if (!relay_log_purge)
-    rli->log_space_limit= 0;
-  pthread_cond_broadcast(&rli->data_cond);
-
-  pthread_mutex_unlock(log_lock);
-
-  if (need_data_lock)
-    pthread_mutex_unlock(&rli->data_lock);
-  if (!rli->relay_log.description_event_for_exec->is_valid() && !*errmsg)
-    *errmsg= "Invalid Format_description log event; could be out of memory";
-
-  DBUG_RETURN ((*errmsg) ? 1 : 0);
-}
-
-
-/*
   Init function to set up array for errors that should be skipped for slave
 
   SYNOPSIS
@@ -461,174 +245,6 @@ void init_slave_skip_errors(const char* arg)
 }
 
 
-void st_relay_log_info::inc_group_relay_log_pos(ulonglong log_pos,
-                                                bool skip_lock)
-{
-  DBUG_ENTER("st_relay_log_info::inc_group_relay_log_pos");
-
-  if (!skip_lock)
-    pthread_mutex_lock(&data_lock);
-  inc_event_relay_log_pos();
-  group_relay_log_pos= event_relay_log_pos;
-  strmake(group_relay_log_name,event_relay_log_name,
-          sizeof(group_relay_log_name)-1);
-
-  notify_group_relay_log_name_update();
-
-  /*
-    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).
-  */
-  /*
-    In 4.x we used the event's len to compute the positions here. This is
-    wrong if the event was 3.23/4.0 and has been converted to 5.0, because
-    then the event's len is not what is was in the master's binlog, so this
-    will make a wrong group_master_log_pos (yes it's a bug in 3.23->4.0
-    replication: Exec_master_log_pos is wrong). Only way to solve this is to
-    have the original offset of the end of the event the relay log. This is
-    what we do in 5.0: log_pos has become "end_log_pos" (because the real use
-    of log_pos in 4.0 was to compute the end_log_pos; so better to store
-    end_log_pos instead of begin_log_pos.
-    If we had not done this fix here, the problem would also have appeared
-    when the slave and master are 5.0 but with different event length (for
-    example the slave is more recent than the master and features the event
-    UID). It would give false MASTER_POS_WAIT, false Exec_master_log_pos in
-    SHOW SLAVE STATUS, and so the user would do some CHANGE MASTER using this
-    value which would lead to badly broken replication.
-    Even the relay_log_pos will be corrupted in this case, because the len is
-    the relay log is not "val".
-    With the end_log_pos solution, we avoid computations involving lengthes.
-  */
-  DBUG_PRINT("info", ("log_pos: %lu  group_master_log_pos: %lu",
-                      (long) log_pos, (long) group_master_log_pos));
-  if (log_pos) // 3.23 binlogs don't have log_posx
-  {
-    group_master_log_pos= log_pos;
-  }
-  pthread_cond_broadcast(&data_cond);
-  if (!skip_lock)
-    pthread_mutex_unlock(&data_lock);
-  DBUG_VOID_RETURN;
-}
-
-
-void st_relay_log_info::close_temporary_tables()
-{
-  TABLE *table,*next;
-  DBUG_ENTER("st_relay_log_info::close_temporary_tables");
-
-  for (table=save_temporary_tables ; table ; table=next)
-  {
-    next=table->next;
-    /*
-      Don't ask for disk deletion. For now, anyway they will be deleted when
-      slave restarts, but it is a better intention to not delete them.
-    */
-    DBUG_PRINT("info", ("table: %p", table));
-    close_temporary(table, 1, 0);
-  }
-  save_temporary_tables= 0;
-  slave_open_temp_tables= 0;
-  DBUG_VOID_RETURN;
-}
-
-/*
-  purge_relay_logs()
-
-  NOTES
-    Assumes to have a run lock on rli and that no slave thread are running.
-*/
-
-int purge_relay_logs(RELAY_LOG_INFO* rli, THD *thd, bool just_reset,
-                     const char** errmsg)
-{
-  int error=0;
-  DBUG_ENTER("purge_relay_logs");
-
-  /*
-    Even if rli->inited==0, we still try to empty rli->master_log_* variables.
-    Indeed, rli->inited==0 does not imply that they already are empty.
-    It could be that slave's info initialization partly succeeded :
-    for example if relay-log.info existed but *relay-bin*.*
-    have been manually removed, init_relay_log_info reads the old
-    relay-log.info and fills rli->master_log_*, then init_relay_log_info
-    checks for the existence of the relay log, this fails and
-    init_relay_log_info leaves rli->inited to 0.
-    In that pathological case, rli->master_log_pos* will be properly reinited
-    at the next START SLAVE (as RESET SLAVE or CHANGE
-    MASTER, the callers of purge_relay_logs, will delete bogus *.info files
-    or replace them with correct files), however if the user does SHOW SLAVE
-    STATUS before START SLAVE, he will see old, confusing rli->master_log_*.
-    In other words, we reinit rli->master_log_* for SHOW SLAVE STATUS
-    to display fine in any case.
-  */
-
-  rli->group_master_log_name[0]= 0;
-  rli->group_master_log_pos= 0;
-
-  if (!rli->inited)
-  {
-    DBUG_PRINT("info", ("rli->inited == 0"));
-    DBUG_RETURN(0);
-  }
-
-  DBUG_ASSERT(rli->slave_running == 0);
-  DBUG_ASSERT(rli->mi->slave_running == 0);
-
-  rli->slave_skip_counter=0;
-  pthread_mutex_lock(&rli->data_lock);
-
-  /*
-    we close the relay log fd possibly left open by the slave SQL thread,
-    to be able to delete it; the relay log fd possibly left open by the slave
-    I/O thread will be closed naturally in reset_logs() by the
-    close(LOG_CLOSE_TO_BE_OPENED) call
-  */
-  if (rli->cur_log_fd >= 0)
-  {
-    end_io_cache(&rli->cache_buf);
-    my_close(rli->cur_log_fd, MYF(MY_WME));
-    rli->cur_log_fd= -1;
-  }
-
-  if (rli->relay_log.reset_logs(thd))
-  {
-    *errmsg = "Failed during log reset";
-    error=1;
-    goto err;
-  }
-  /* Save name of used relay log file */
-  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);
-  rli->group_relay_log_pos= rli->event_relay_log_pos= BIN_LOG_HEADER_SIZE;
-  if (count_relay_log_space(rli))
-  {
-    *errmsg= "Error counting relay log space";
-    goto err;
-  }
-  if (!just_reset)
-    error= init_relay_log_pos(rli, rli->group_relay_log_name,
-                              rli->group_relay_log_pos,
-                              0 /* do not need data lock */, errmsg, 0);
-
-err:
-#ifndef DBUG_OFF
-  char buf[22];
-#endif
-  DBUG_PRINT("info",("log_space_total: %s",llstr(rli->log_space_total,buf)));
-  pthread_mutex_unlock(&rli->data_lock);
-  DBUG_RETURN(error);
-}
-
-
 int terminate_slave_threads(MASTER_INFO* mi,int thread_mask,bool skip_lock)
 {
   DBUG_ENTER("terminate_slave_threads");
@@ -1023,7 +639,7 @@ const char *print_slave_db_safe(const char* db)
   DBUG_RETURN((db ? db : ""));
 }
 
-static int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
+int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
                                  const char *default_val)
 {
   uint length;
@@ -1054,7 +670,7 @@ static int init_strvar_from_file(char *var, int max_size, IO_CACHE *f,
 }
 
 
-static int init_intvar_from_file(int* var, IO_CACHE* f, int default_val)
+int init_intvar_from_file(int* var, IO_CACHE* f, int default_val)
 {
   char buf[32];
   DBUG_ENTER("init_intvar_from_file");
@@ -1456,261 +1072,6 @@ int fetch_master_table(THD *thd, const char *db_name, const char *table_name,
 }
 
 
-void end_master_info(MASTER_INFO* mi)
-{
-  DBUG_ENTER("end_master_info");
-
-  if (!mi->inited)
-    DBUG_VOID_RETURN;
-  end_relay_log_info(&mi->rli);
-  if (mi->fd >= 0)
-  {
-    end_io_cache(&mi->file);
-    (void)my_close(mi->fd, MYF(MY_WME));
-    mi->fd = -1;
-  }
-  mi->inited = 0;
-
-  DBUG_VOID_RETURN;
-}
-
-
-static int init_relay_log_info(RELAY_LOG_INFO* rli,
-                               const char* info_fname)
-{
-  char fname[FN_REFLEN+128];
-  int info_fd;
-  const char* msg = 0;
-  int error = 0;
-  DBUG_ENTER("init_relay_log_info");
-  DBUG_ASSERT(!rli->no_storage);         // Don't init if there is no storage
-
-  if (rli->inited)                       // Set if this function called
-    DBUG_RETURN(0);
-  fn_format(fname, info_fname, mysql_data_home, "", 4+32);
-  pthread_mutex_lock(&rli->data_lock);
-  info_fd = rli->info_fd;
-  rli->cur_log_fd = -1;
-  rli->slave_skip_counter=0;
-  rli->abort_pos_wait=0;
-  rli->log_space_limit= relay_log_space_limit;
-  rli->log_space_total= 0;
-  rli->tables_to_lock= 0;
-  rli->tables_to_lock_count= 0;
-
-  /*
-    The relay log will now be opened, as a SEQ_READ_APPEND IO_CACHE.
-    Note that the I/O thread flushes it to disk after writing every
-    event, in flush_master_info(mi, 1).
-  */
-
-  /*
-    For the maximum log size, we choose max_relay_log_size if it is
-    non-zero, max_binlog_size otherwise. If later the user does SET
-    GLOBAL on one of these variables, fix_max_binlog_size and
-    fix_max_relay_log_size will reconsider the choice (for example
-    if the user changes max_relay_log_size to zero, we have to
-    switch to using max_binlog_size for the relay log) and update
-    rli->relay_log.max_size (and mysql_bin_log.max_size).
-  */
-  {
-    char buf[FN_REFLEN];
-    const char *ln;
-    static bool name_warning_sent= 0;
-    ln= rli->relay_log.generate_name(opt_relay_logname, "-relay-bin",
-                                     1, buf);
-    /* We send the warning only at startup, not after every RESET SLAVE */
-    if (!opt_relay_logname && !opt_relaylog_index_name && !name_warning_sent)
-    {
-      /*
-        User didn't give us info to name the relay log index file.
-        Picking `hostname`-relay-bin.index like we do, causes replication to
-        fail if this slave's hostname is changed later. So, we would like to
-        instead require a name. But as we don't want to break many existing
-        setups, we only give warning, not error.
-      */
-      sql_print_warning("Neither --relay-log nor --relay-log-index were used;"
-                        " so replication "
-                        "may break when this MySQL server acts as a "
-                        "slave and has his hostname changed!! Please "
-                        "use '--relay-log=%s' to avoid this problem.", ln);
-      name_warning_sent= 1;
-    }
-    /*
-      note, that if open() fails, we'll still have index file open
-      but a destructor will take care of that
-    */
-    if (rli->relay_log.open_index_file(opt_relaylog_index_name, ln) ||
-        rli->relay_log.open(ln, LOG_BIN, 0, SEQ_READ_APPEND, 0,
-                            (max_relay_log_size ? max_relay_log_size :
-                            max_binlog_size), 1))
-    {
-      pthread_mutex_unlock(&rli->data_lock);
-      sql_print_error("Failed in open_log() called from init_relay_log_info()");
-      DBUG_RETURN(1);
-    }
-  }
-
-  /* if file does not exist */
-  if (access(fname,F_OK))
-  {
-    /*
-      If someone removed the file from underneath our feet, just close
-      the old descriptor and re-create the old file
-    */
-    if (info_fd >= 0)
-      my_close(info_fd, MYF(MY_WME));
-    if ((info_fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0)
-    {
-      sql_print_error("Failed to create a new relay log info file (\
-file '%s', errno %d)", fname, my_errno);
-      msg= current_thd->net.last_error;
-      goto err;
-    }
-    if (init_io_cache(&rli->info_file, info_fd, IO_SIZE*2, READ_CACHE, 0L,0,
-                      MYF(MY_WME)))
-    {
-      sql_print_error("Failed to create a cache on relay log info file '%s'",
-                      fname);
-      msg= current_thd->net.last_error;
-      goto err;
-    }
-
-    /* Init relay log with first entry in the relay index file */
-    if (init_relay_log_pos(rli,NullS,BIN_LOG_HEADER_SIZE,0 /* no data lock */,
-                           &msg, 0))
-    {
-      sql_print_error("Failed to open the relay log 'FIRST' (relay_log_pos 4)");
-      goto err;
-    }
-    rli->group_master_log_name[0]= 0;
-    rli->group_master_log_pos= 0;
-    rli->info_fd= info_fd;
-  }
-  else // file exists
-  {
-    if (info_fd >= 0)
-      reinit_io_cache(&rli->info_file, READ_CACHE, 0L,0,0);
-    else
-    {
-      int error=0;
-      if ((info_fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0)
-      {
-        sql_print_error("\
-Failed to open the existing relay log info file '%s' (errno %d)",
-                        fname, my_errno);
-        error= 1;
-      }
-      else if (init_io_cache(&rli->info_file, info_fd,
-                             IO_SIZE*2, READ_CACHE, 0L, 0, MYF(MY_WME)))
-      {
-        sql_print_error("Failed to create a cache on relay log info file '%s'",
-                        fname);
-        error= 1;
-      }
-      if (error)
-      {
-        if (info_fd >= 0)
-          my_close(info_fd, MYF(0));
-        rli->info_fd= -1;
-        rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
-        pthread_mutex_unlock(&rli->data_lock);
-        DBUG_RETURN(1);
-      }
-    }
-
-    rli->info_fd = info_fd;
-    int relay_log_pos, master_log_pos;
-    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->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;
-    }
-    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->group_relay_log_name,
-                           rli->group_relay_log_pos,
-                           0 /* no data lock*/,
-                           &msg, 0))
-    {
-      char llbuf[22];
-      sql_print_error("Failed to open the relay log '%s' (relay_log_pos %s)",
-                      rli->group_relay_log_name,
-                      llstr(rli->group_relay_log_pos, llbuf));
-      goto err;
-    }
-  }
-
-#ifndef DBUG_OFF
-  {
-    char llbuf1[22], llbuf2[22];
-    DBUG_PRINT("info", ("my_b_tell(rli->cur_log)=%s rli->event_relay_log_pos=%s",
-                        llstr(my_b_tell(rli->cur_log),llbuf1),
-                        llstr(rli->event_relay_log_pos,llbuf2)));
-    DBUG_ASSERT(rli->event_relay_log_pos >= BIN_LOG_HEADER_SIZE);
-    DBUG_ASSERT(my_b_tell(rli->cur_log) == rli->event_relay_log_pos);
-  }
-#endif
-
-  /*
-    Now change the cache from READ to WRITE - must do this
-    before flush_relay_log_info
-  */
-  reinit_io_cache(&rli->info_file, WRITE_CACHE,0L,0,1);
-  if ((error= flush_relay_log_info(rli)))
-    sql_print_error("Failed to flush relay log info file");
-  if (count_relay_log_space(rli))
-  {
-    msg="Error counting relay log space";
-    goto err;
-  }
-  rli->inited= 1;
-  pthread_mutex_unlock(&rli->data_lock);
-  DBUG_RETURN(error);
-
-err:
-  sql_print_error(msg);
-  end_io_cache(&rli->info_file);
-  if (info_fd >= 0)
-    my_close(info_fd, MYF(0));
-  rli->info_fd= -1;
-  rli->relay_log.close(LOG_CLOSE_INDEX | LOG_CLOSE_STOP_EVENT);
-  pthread_mutex_unlock(&rli->data_lock);
-  DBUG_RETURN(1);
-}
-
-
-static inline int add_relay_log(RELAY_LOG_INFO* rli,LOG_INFO* linfo)
-{
-  MY_STAT s;
-  DBUG_ENTER("add_relay_log");
-  if (!my_stat(linfo->log_file_name,&s,MYF(0)))
-  {
-    sql_print_error("log %s listed in the index, but failed to stat",
-                    linfo->log_file_name);
-    DBUG_RETURN(1);
-  }
-  rli->log_space_total += s.st_size;
-#ifndef DBUG_OFF
-  char buf[22];
-  DBUG_PRINT("info",("log_space_total: %s", llstr(rli->log_space_total,buf)));
-#endif
-  DBUG_RETURN(0);
-}
-
-
 static bool wait_for_relay_log_space(RELAY_LOG_INFO* rli)
 {
   bool slave_killed=0;
@@ -1733,31 +1094,6 @@ Waiting for the slave SQL thread to free enough relay log space");
 }
 
 
-static int count_relay_log_space(RELAY_LOG_INFO* rli)
-{
-  LOG_INFO linfo;
-  DBUG_ENTER("count_relay_log_space");
-  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");
-    DBUG_RETURN(1);
-  }
-  do
-  {
-    if (add_relay_log(rli,&linfo))
-      DBUG_RETURN(1);
-  } while (!rli->relay_log.find_next_log(&linfo, 1));
-  /*
-     As we have counted everything, including what may have written in a
-     preceding write, we must reset bytes_written, or we may count some space
-     twice.
-  */
-  rli->relay_log.reset_bytes_written();
-  DBUG_RETURN(0);
-}
-
-
 /*
   Builds a Rotate from the ignored events' info and writes it to relay log.
 
@@ -1811,284 +1147,6 @@ static void write_ignored_events_info_to_relay_log(THD *thd, MASTER_INFO *mi)
 }
 
 
-void init_master_info_with_options(MASTER_INFO* mi)
-{
-  DBUG_ENTER("init_master_info_with_options");
-
-  mi->master_log_name[0] = 0;
-  mi->master_log_pos = BIN_LOG_HEADER_SIZE;             // skip magic number
-
-  if (master_host)
-    strmake(mi->host, master_host, sizeof(mi->host) - 1);
-  if (master_user)
-    strmake(mi->user, master_user, sizeof(mi->user) - 1);
-  if (master_password)
-    strmake(mi->password, master_password, MAX_PASSWORD_LENGTH);
-  mi->port = master_port;
-  mi->connect_retry = master_connect_retry;
-
-  mi->ssl= master_ssl;
-  if (master_ssl_ca)
-    strmake(mi->ssl_ca, master_ssl_ca, sizeof(mi->ssl_ca)-1);
-  if (master_ssl_capath)
-    strmake(mi->ssl_capath, master_ssl_capath, sizeof(mi->ssl_capath)-1);
-  if (master_ssl_cert)
-    strmake(mi->ssl_cert, master_ssl_cert, sizeof(mi->ssl_cert)-1);
-  if (master_ssl_cipher)
-    strmake(mi->ssl_cipher, master_ssl_cipher, sizeof(mi->ssl_cipher)-1);
-  if (master_ssl_key)
-    strmake(mi->ssl_key, master_ssl_key, sizeof(mi->ssl_key)-1);
-  DBUG_VOID_RETURN;
-}
-
-void clear_slave_error(RELAY_LOG_INFO* rli)
-{
-  DBUG_ENTER("clear_slave_error");
-
-  /* Clear the errors displayed by SHOW SLAVE STATUS */
-  rli->last_slave_error[0]= 0;
-  rli->last_slave_errno= 0;
-  DBUG_VOID_RETURN;
-}
-
-/*
-    Reset UNTIL condition for RELAY_LOG_INFO
-   SYNOPSYS
-    clear_until_condition()
-      rli - RELAY_LOG_INFO structure where UNTIL condition should be reset
- */
-void clear_until_condition(RELAY_LOG_INFO* rli)
-{
-  DBUG_ENTER("clear_until_condition");
-
-  rli->until_condition= RELAY_LOG_INFO::UNTIL_NONE;
-  rli->until_log_name[0]= 0;
-  rli->until_log_pos= 0;
-  DBUG_VOID_RETURN;
-}
-
-
-#define LINES_IN_MASTER_INFO_WITH_SSL 14
-
-
-int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
-                     const char* slave_info_fname,
-                     bool abort_if_no_master_info_file,
-                     int thread_mask)
-{
-  int fd,error;
-  char fname[FN_REFLEN+128];
-  DBUG_ENTER("init_master_info");
-
-  if (mi->inited)
-  {
-    /*
-      We have to reset read position of relay-log-bin as we may have
-      already been reading from 'hotlog' when the slave was stopped
-      last time. If this case pos_in_file would be set and we would
-      get a crash when trying to read the signature for the binary
-      relay log.
-
-      We only rewind the read position if we are starting the SQL
-      thread. The handle_slave_sql thread assumes that the read
-      position is at the beginning of the file, and will read the
-      "signature" and then fast-forward to the last position read.
-    */
-    if (thread_mask & SLAVE_SQL)
-    {
-      my_b_seek(mi->rli.cur_log, (my_off_t) 0);
-    }
-    DBUG_RETURN(0);
-  }
-
-  mi->mysql=0;
-  mi->file_id=1;
-  fn_format(fname, master_info_fname, mysql_data_home, "", 4+32);
-
-  /*
-    We need a mutex while we are changing master info parameters to
-    keep other threads from reading bogus info
-  */
-
-  pthread_mutex_lock(&mi->data_lock);
-  fd = mi->fd;
-
-  /* does master.info exist ? */
-
-  if (access(fname,F_OK))
-  {
-    if (abort_if_no_master_info_file)
-    {
-      pthread_mutex_unlock(&mi->data_lock);
-      DBUG_RETURN(0);
-    }
-    /*
-      if someone removed the file from underneath our feet, just close
-      the old descriptor and re-create the old file
-    */
-    if (fd >= 0)
-      my_close(fd, MYF(MY_WME));
-    if ((fd = my_open(fname, O_CREAT|O_RDWR|O_BINARY, MYF(MY_WME))) < 0 )
-    {
-      sql_print_error("Failed to create a new master info file (\
-file '%s', errno %d)", fname, my_errno);
-      goto err;
-    }
-    if (init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,0,
-                      MYF(MY_WME)))
-    {
-      sql_print_error("Failed to create a cache on master info file (\
-file '%s')", fname);
-      goto err;
-    }
-
-    mi->fd = fd;
-    init_master_info_with_options(mi);
-
-  }
-  else // file exists
-  {
-    if (fd >= 0)
-      reinit_io_cache(&mi->file, READ_CACHE, 0L,0,0);
-    else
-    {
-      if ((fd = my_open(fname, O_RDWR|O_BINARY, MYF(MY_WME))) < 0 )
-      {
-        sql_print_error("Failed to open the existing master info file (\
-file '%s', errno %d)", fname, my_errno);
-        goto err;
-      }
-      if (init_io_cache(&mi->file, fd, IO_SIZE*2, READ_CACHE, 0L,
-                        0, MYF(MY_WME)))
-      {
-        sql_print_error("Failed to create a cache on master info file (\
-file '%s')", fname);
-        goto err;
-      }
-    }
-
-    mi->fd = fd;
-    int port, connect_retry, master_log_pos, ssl= 0, lines;
-    char *first_non_digit;
-
-    /*
-       Starting from 4.1.x master.info has new format. Now its
-       first line contains number of lines in file. By reading this
-       number we will be always distinguish to which version our
-       master.info corresponds to. We can't simply count lines in
-       file since versions before 4.1.x could generate files with more
-       lines than needed.
-       If first line doesn't contain a number or contain number less than
-       14 then such file is treated like file from pre 4.1.1 version.
-       There is no ambiguity when reading an old master.info, as before
-       4.1.1, the first line contained the binlog's name, which is either
-       empty or has an extension (contains a '.'), so can't be confused
-       with an integer.
-
-       So we're just reading first line and trying to figure which version
-       is this.
-    */
-
-    /*
-       The first row is temporarily stored in mi->master_log_name,
-       if it is line count and not binlog name (new format) it will be
-       overwritten by the second row later.
-    */
-    if (init_strvar_from_file(mi->master_log_name,
-                              sizeof(mi->master_log_name), &mi->file,
-                              ""))
-      goto errwithmsg;
-
-    lines= strtoul(mi->master_log_name, &first_non_digit, 10);
-
-    if (mi->master_log_name[0]!='\0' &&
-        *first_non_digit=='\0' && lines >= LINES_IN_MASTER_INFO_WITH_SSL)
-    {                                          // Seems to be new format
-      if (init_strvar_from_file(mi->master_log_name,
-            sizeof(mi->master_log_name), &mi->file, ""))
-        goto errwithmsg;
-    }
-    else
-      lines= 7;
-
-    if (init_intvar_from_file(&master_log_pos, &mi->file, 4) ||
-        init_strvar_from_file(mi->host, sizeof(mi->host), &mi->file,
-                              master_host) ||
-        init_strvar_from_file(mi->user, sizeof(mi->user), &mi->file,
-                              master_user) ||
-        init_strvar_from_file(mi->password, SCRAMBLED_PASSWORD_CHAR_LENGTH+1,
-                              &mi->file, master_password) ||
-        init_intvar_from_file(&port, &mi->file, master_port) ||
-        init_intvar_from_file(&connect_retry, &mi->file,
-                              master_connect_retry))
-      goto errwithmsg;
-
-    /*
-       If file has ssl part use it even if we have server without
-       SSL support. But these option will be ignored later when
-       slave will try connect to master, so in this case warning
-       is printed.
-     */
-    if (lines >= LINES_IN_MASTER_INFO_WITH_SSL &&
-        (init_intvar_from_file(&ssl, &mi->file, master_ssl) ||
-         init_strvar_from_file(mi->ssl_ca, sizeof(mi->ssl_ca),
-                               &mi->file, master_ssl_ca) ||
-         init_strvar_from_file(mi->ssl_capath, sizeof(mi->ssl_capath),
-                               &mi->file, master_ssl_capath) ||
-         init_strvar_from_file(mi->ssl_cert, sizeof(mi->ssl_cert),
-                               &mi->file, master_ssl_cert) ||
-         init_strvar_from_file(mi->ssl_cipher, sizeof(mi->ssl_cipher),
-                               &mi->file, master_ssl_cipher) ||
-         init_strvar_from_file(mi->ssl_key, sizeof(mi->ssl_key),
-                              &mi->file, master_ssl_key)))
-      goto errwithmsg;
-#ifndef HAVE_OPENSSL
-    if (ssl)
-      sql_print_warning("SSL information in the master info file "
-                      "('%s') are ignored because this MySQL slave was compiled "
-                      "without SSL support.", fname);
-#endif /* HAVE_OPENSSL */
-
-    /*
-      This has to be handled here as init_intvar_from_file can't handle
-      my_off_t types
-    */
-    mi->master_log_pos= (my_off_t) master_log_pos;
-    mi->port= (uint) port;
-    mi->connect_retry= (uint) connect_retry;
-    mi->ssl= (my_bool) ssl;
-  }
-  DBUG_PRINT("master_info",("log_file_name: %s  position: %ld",
-                            mi->master_log_name,
-                            (ulong) mi->master_log_pos));
-
-  mi->rli.mi = mi;
-  if (init_relay_log_info(&mi->rli, slave_info_fname))
-    goto err;
-
-  mi->inited = 1;
-  // now change cache READ -> WRITE - must do this before flush_master_info
-  reinit_io_cache(&mi->file, WRITE_CACHE, 0L, 0, 1);
-  if ((error=test(flush_master_info(mi, 1))))
-    sql_print_error("Failed to flush master info file");
-  pthread_mutex_unlock(&mi->data_lock);
-  DBUG_RETURN(error);
-
-errwithmsg:
-  sql_print_error("Error reading master configuration");
-
-err:
-  if (fd >= 0)
-  {
-    my_close(fd, MYF(0));
-    end_io_cache(&mi->file);
-  }
-  mi->fd= -1;
-  pthread_mutex_unlock(&mi->data_lock);
-  DBUG_RETURN(1);
-}
-
-
 int register_slave_on_master(MYSQL* mysql)
 {
   char buf[1024], *pos= buf;
@@ -2307,313 +1365,6 @@ bool show_master_info(THD* thd, MASTER_INFO* mi)
   DBUG_RETURN(FALSE);
 }
 
-/*
-  RETURN
-     2 - flush relay log failed
-     1 - flush master info failed
-     0 - all ok
-*/
-int flush_master_info(MASTER_INFO* mi, bool flush_relay_log_cache)
-{
-  IO_CACHE* file = &mi->file;
-  char lbuf[22];
-  DBUG_ENTER("flush_master_info");
-  DBUG_PRINT("enter",("master_pos: %ld", (long) mi->master_log_pos));
-
-  /*
-    Flush the relay log to disk. If we don't do it, then the relay log while
-    have some part (its last kilobytes) in memory only, so if the slave server
-    dies now, with, say, from master's position 100 to 150 in memory only (not
-    on disk), and with position 150 in master.info, then when the slave
-    restarts, the I/O thread will fetch binlogs from 150, so in the relay log
-    we will have "[0, 100] U [150, infinity[" and nobody will notice it, so the
-    SQL thread will jump from 100 to 150, and replication will silently break.
-
-    When we come to this place in code, relay log may or not be initialized;
-    the caller is responsible for setting 'flush_relay_log_cache' accordingly.
-  */
-  if (flush_relay_log_cache &&
-      flush_io_cache(mi->rli.relay_log.get_log_file()))
-    DBUG_RETURN(2);
-
-  /*
-    We flushed the relay log BEFORE the master.info file, because if we crash
-    now, we will get a duplicate event in the relay log at restart. If we
-    flushed in the other order, we would get a hole in the relay log.
-    And duplicate is better than hole (with a duplicate, in later versions we
-    can add detection and scrap one event; with a hole there's nothing we can
-    do).
-  */
-
-  /*
-     In certain cases this code may create master.info files that seems
-     corrupted, because of extra lines filled with garbage in the end
-     file (this happens if new contents take less space than previous
-     contents of file). But because of number of lines in the first line
-     of file we don't care about this garbage.
-  */
-
-  my_b_seek(file, 0L);
-  my_b_printf(file, "%u\n%s\n%s\n%s\n%s\n%s\n%d\n%d\n%d\n%s\n%s\n%s\n%s\n%s\n",
-              LINES_IN_MASTER_INFO_WITH_SSL,
-              mi->master_log_name, llstr(mi->master_log_pos, lbuf),
-              mi->host, mi->user,
-              mi->password, mi->port, mi->connect_retry,
-              (int)(mi->ssl), mi->ssl_ca, mi->ssl_capath, mi->ssl_cert,
-              mi->ssl_cipher, mi->ssl_key);
-  DBUG_RETURN(-flush_io_cache(file));
-}
-
-
-st_relay_log_info::st_relay_log_info()
-  :no_storage(FALSE), 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), last_master_timestamp(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), until_condition(UNTIL_NONE),
-   until_log_pos(0), retried_trans(0),
-   tables_to_lock(0), tables_to_lock_count(0),
-   m_reload_flags(RELOAD_NONE_F),
-   unsafe_to_stop_at(0)
-{
-  DBUG_ENTER("st_relay_log_info::st_relay_log_info");
-
-  group_relay_log_name[0]= event_relay_log_name[0]=
-    group_master_log_name[0]= 0;
-  last_slave_error[0]= until_log_name[0]= ign_master_log_name_end[0]= 0;
-  bzero((char*) &info_file, sizeof(info_file));
-  bzero((char*) &cache_buf, sizeof(cache_buf));
-  cached_charset_invalidate();
-  pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
-  pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
-  pthread_mutex_init(&log_space_lock, MY_MUTEX_INIT_FAST);
-  pthread_cond_init(&data_cond, NULL);
-  pthread_cond_init(&start_cond, NULL);
-  pthread_cond_init(&stop_cond, NULL);
-  pthread_cond_init(&log_space_cond, NULL);
-  relay_log.init_pthread_objects();
-  DBUG_VOID_RETURN;
-}
-
-
-st_relay_log_info::~st_relay_log_info()
-{
-  DBUG_ENTER("st_relay_log_info::~st_relay_log_info");
-
-  pthread_mutex_destroy(&run_lock);
-  pthread_mutex_destroy(&data_lock);
-  pthread_mutex_destroy(&log_space_lock);
-  pthread_cond_destroy(&data_cond);
-  pthread_cond_destroy(&start_cond);
-  pthread_cond_destroy(&stop_cond);
-  pthread_cond_destroy(&log_space_cond);
-  relay_log.cleanup();
-  DBUG_VOID_RETURN;
-}
-
-/*
-  Waits until the SQL thread reaches (has executed up to) the
-  log/position or timed out.
-
-  SYNOPSIS
-    wait_for_pos()
-    thd             client thread that sent SELECT MASTER_POS_WAIT
-    log_name        log name to wait for
-    log_pos         position to wait for
-    timeout         timeout in seconds before giving up waiting
-
-  NOTES
-    timeout is longlong whereas it should be ulong ; but this is
-    to catch if the user submitted a negative timeout.
-
-  RETURN VALUES
-    -2          improper arguments (log_pos<0)
-                or slave not running, or master info changed
-                during the function's execution,
-                or client thread killed. -2 is translated to NULL by caller
-    -1          timed out
-    >=0         number of log events the function had to wait
-                before reaching the desired log/position
- */
-
-int st_relay_log_info::wait_for_pos(THD* thd, String* log_name,
-                                    longlong log_pos,
-                                    longlong timeout)
-{
-  int event_count = 0;
-  ulong init_abort_pos_wait;
-  int error=0;
-  struct timespec abstime; // for timeout checking
-  const char *msg;
-  DBUG_ENTER("st_relay_log_info::wait_for_pos");
-
-  if (!inited)
-    DBUG_RETURN(-1);
-
-  DBUG_PRINT("enter",("log_name: '%s'  log_pos: %lu  timeout: %lu",
-                      log_name->c_ptr(), (ulong) log_pos, (ulong) timeout));
-
-  set_timespec(abstime,timeout);
-  pthread_mutex_lock(&data_lock);
-  msg= thd->enter_cond(&data_cond, &data_lock,
-                       "Waiting for the slave SQL thread to "
-                       "advance position");
-  /*
-     This function will abort when it notices that some CHANGE MASTER or
-     RESET MASTER has changed the master info.
-     To catch this, these commands modify abort_pos_wait ; We just monitor
-     abort_pos_wait and see if it has changed.
-     Why do we have this mechanism instead of simply monitoring slave_running
-     in the loop (we do this too), as CHANGE MASTER/RESET SLAVE require that
-     the SQL thread be stopped?
-     This is becasue if someones does:
-     STOP SLAVE;CHANGE MASTER/RESET SLAVE; START SLAVE;
-     the change may happen very quickly and we may not notice that
-     slave_running briefly switches between 1/0/1.
-  */
-  init_abort_pos_wait= abort_pos_wait;
-
-  /*
-    We'll need to
-    handle all possible log names comparisons (e.g. 999 vs 1000).
-    We use ulong for string->number conversion ; this is no
-    stronger limitation than in find_uniq_filename in sql/log.cc
-  */
-  ulong log_name_extension;
-  char log_name_tmp[FN_REFLEN]; //make a char[] from String
-
-  strmake(log_name_tmp, log_name->ptr(), min(log_name->length(), FN_REFLEN-1));
-
-  char *p= fn_ext(log_name_tmp);
-  char *p_end;
-  if (!*p || log_pos<0)
-  {
-    error= -2; //means improper arguments
-    goto err;
-  }
-  // Convert 0-3 to 4
-  log_pos= max(log_pos, BIN_LOG_HEADER_SIZE);
-  /* p points to '.' */
-  log_name_extension= strtoul(++p, &p_end, 10);
-  /*
-    p_end points to the first invalid character.
-    If it equals to p, no digits were found, error.
-    If it contains '\0' it means conversion went ok.
-  */
-  if (p_end==p || *p_end)
-  {
-    error= -2;
-    goto err;
-  }
-
-  /* The "compare and wait" main loop */
-  while (!thd->killed &&
-         init_abort_pos_wait == abort_pos_wait &&
-         slave_running)
-  {
-    bool pos_reached;
-    int cmp_result= 0;
-
-    DBUG_PRINT("info",
-               ("init_abort_pos_wait: %ld  abort_pos_wait: %ld",
-                init_abort_pos_wait, abort_pos_wait));
-    DBUG_PRINT("info",("group_master_log_name: '%s'  pos: %lu",
-                       group_master_log_name, (ulong) group_master_log_pos));
-
-    /*
-      group_master_log_name can be "", if we are just after a fresh
-      replication start or after a CHANGE MASTER TO MASTER_HOST/PORT
-      (before we have executed one Rotate event from the master) or
-      (rare) if the user is doing a weird slave setup (see next
-      paragraph).  If group_master_log_name is "", we assume we don't
-      have enough info to do the comparison yet, so we just wait until
-      more data. In this case master_log_pos is always 0 except if
-      somebody (wrongly) sets this slave to be a slave of itself
-      without using --replicate-same-server-id (an unsupported
-      configuration which does nothing), then group_master_log_pos
-      will grow and group_master_log_name will stay "".
-    */
-    if (*group_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,
-        and protect against user's input error :
-        if the names do not match up to '.' included, return error
-      */
-      char *q= (char*)(fn_ext(basename)+1);
-      if (strncmp(basename, log_name_tmp, (int)(q-basename)))
-      {
-        error= -2;
-        break;
-      }
-      // Now compare extensions.
-      char *q_end;
-      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= (group_master_log_name_extension > log_name_extension) ? 1 : 0 ;
-
-      pos_reached= ((!cmp_result && group_master_log_pos >= (ulonglong)log_pos) ||
-                    cmp_result > 0);
-      if (pos_reached || thd->killed)
-        break;
-    }
-
-    //wait for master update, with optional timeout.
-
-    DBUG_PRINT("info",("Waiting for master update"));
-    /*
-      We are going to pthread_cond_(timed)wait(); if the SQL thread stops it
-      will wake us up.
-    */
-    if (timeout > 0)
-    {
-      /*
-        Note that pthread_cond_timedwait checks for the timeout
-        before for the condition ; i.e. it returns ETIMEDOUT
-        if the system time equals or exceeds the time specified by abstime
-        before the condition variable is signaled or broadcast, _or_ if
-        the absolute time specified by abstime has already passed at the time
-        of the call.
-        For that reason, pthread_cond_timedwait will do the "timeoutting" job
-        even if its condition is always immediately signaled (case of a loaded
-        master).
-      */
-      error=pthread_cond_timedwait(&data_cond, &data_lock, &abstime);
-    }
-    else
-      pthread_cond_wait(&data_cond, &data_lock);
-    DBUG_PRINT("info",("Got signal of master update or timed out"));
-    if (error == ETIMEDOUT || error == ETIME)
-    {
-      error= -1;
-      break;
-    }
-    error=0;
-    event_count++;
-    DBUG_PRINT("info",("Testing if killed or SQL thread not running"));
-  }
-
-err:
-  thd->exit_cond(msg);
-  DBUG_PRINT("exit",("killed: %d  abort: %d  slave_running: %d \
-improper_arguments: %d  timed_out: %d",
-                     thd->killed_errno(),
-                     (int) (init_abort_pos_wait != abort_pos_wait),
-                     (int) slave_running,
-                     (int) (error == -2),
-                     (int) (error == -1)));
-  if (thd->killed || init_abort_pos_wait != abort_pos_wait ||
-      !slave_running)
-  {
-    error= -2;
-  }
-  DBUG_RETURN( error ? error : event_count );
-}
 
 void set_slave_thread_options(THD* thd)
 {
@@ -2663,6 +1414,13 @@ static int init_slave_thread(THD* thd, SLAVE_THD_TYPE thd_type)
     SYSTEM_THREAD_SLAVE_SQL : SYSTEM_THREAD_SLAVE_IO;
   thd->security_ctx->skip_grants();
   my_net_init(&thd->net, 0);
+/*
+  Adding MAX_LOG_EVENT_HEADER_LEN to the max_allowed_packet on all
+  slave threads, since a replication event can become this much larger
+  than the corresponding packet (query) sent from client to master.
+*/
+  thd->variables.max_allowed_packet= global_system_variables.max_allowed_packet
+    + MAX_LOG_EVENT_HEADER;  /* note, incr over the global not session var */
   thd->net.read_timeout = slave_net_timeout;
   thd->slave_thread = 1;
   set_slave_thread_options(thd);
@@ -2872,119 +1630,6 @@ int check_expected_error(THD* thd, RELAY_LOG_INFO* rli, int expected_error)
   }
 }
 
-/*
-     Check if condition stated in UNTIL clause of START SLAVE is reached.
-   SYNOPSYS
-     st_relay_log_info::is_until_satisfied()
-   DESCRIPTION
-     Checks if UNTIL condition is reached. Uses caching result of last
-     comparison of current log file name and target log file name. So cached
-     value should be invalidated if current log file name changes
-     (see st_relay_log_info::notify_... functions).
-
-     This caching is needed to avoid of expensive string comparisons and
-     strtol() conversions needed for log names comparison. We don't need to
-     compare them each time this function is called, we only need to do this
-     when current log name changes. If we have UNTIL_MASTER_POS condition we
-     need to do this only after Rotate_log_event::exec_event() (which is
-     rare, so caching gives real benifit), and if we have UNTIL_RELAY_POS
-     condition then we should invalidate cached comarison value after
-     inc_group_relay_log_pos() which called for each group of events (so we
-     have some benefit if we have something like queries that use
-     autoincrement or if we have transactions).
-
-     Should be called ONLY if until_condition != UNTIL_NONE !
-   RETURN VALUE
-     true - condition met or error happened (condition seems to have
-            bad log file name)
-     false - condition not met
-*/
-
-bool st_relay_log_info::is_until_satisfied()
-{
-  const char *log_name;
-  ulonglong log_pos;
-  DBUG_ENTER("st_relay_log_info::is_until_satisfied");
-
-  DBUG_ASSERT(until_condition != UNTIL_NONE);
-
-  if (until_condition == UNTIL_MASTER_POS)
-  {
-    log_name= group_master_log_name;
-    log_pos= group_master_log_pos;
-  }
-  else
-  { /* until_condition == UNTIL_RELAY_POS */
-    log_name= group_relay_log_name;
-    log_pos= group_relay_log_pos;
-  }
-
-  if (until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_UNKNOWN)
-  {
-    /*
-      We have no cached comparison results so we should compare log names
-      and cache result.
-      If we are after RESET SLAVE, and the SQL slave thread has not processed
-      any event yet, it could be that group_master_log_name is "". In that case,
-      just wait for more events (as there is no sensible comparison to do).
-    */
-
-    if (*log_name)
-    {
-      const char *basename= log_name + dirname_length(log_name);
-
-      const char *q= (const char*)(fn_ext(basename)+1);
-      if (strncmp(basename, until_log_name, (int)(q-basename)) == 0)
-      {
-        /* Now compare extensions. */
-        char *q_end;
-        ulong log_name_extension= strtoul(q, &q_end, 10);
-        if (log_name_extension < until_log_name_extension)
-          until_log_names_cmp_result= UNTIL_LOG_NAMES_CMP_LESS;
-        else
-          until_log_names_cmp_result=
-            (log_name_extension > until_log_name_extension) ?
-            UNTIL_LOG_NAMES_CMP_GREATER : UNTIL_LOG_NAMES_CMP_EQUAL ;
-      }
-      else
-      {
-        /* Probably error so we aborting */
-        sql_print_error("Slave SQL thread is stopped because UNTIL "
-                        "condition is bad.");
-        DBUG_RETURN(TRUE);
-      }
-    }
-    else
-      DBUG_RETURN(until_log_pos == 0);
-  }
-
-  DBUG_RETURN(((until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_EQUAL &&
-           log_pos >= until_log_pos) ||
-          until_log_names_cmp_result == UNTIL_LOG_NAMES_CMP_GREATER));
-}
-
-
-void st_relay_log_info::cached_charset_invalidate()
-{
-  DBUG_ENTER("st_relay_log_info::cached_charset_invalidate");
-
-  /* Full of zeroes means uninitialized. */
-  bzero(cached_charset, sizeof(cached_charset));
-  DBUG_VOID_RETURN;
-}
-
-
-bool st_relay_log_info::cached_charset_compare(char *charset)
-{
-  DBUG_ENTER("st_relay_log_info::cached_charset_compare");
-
-  if (bcmp(cached_charset, charset, sizeof(cached_charset)))
-  {
-    memcpy(cached_charset, charset, sizeof(cached_charset));
-    DBUG_RETURN(1);
-  }
-  DBUG_RETURN(0);
-}
 
 /*
   Check if the current error is of temporary nature of not.
@@ -3176,6 +1821,8 @@ static int exec_relay_log_event(THD* thd, RELAY_LOG_INFO* rli)
           We were in a transaction which has been rolled back because of a
           temporary error;
           let's seek back to BEGIN log event and retry it all again.
+	  Note, if lock wait timeout (innodb_lock_wait_timeout exceeded)
+	  there is no rollback since 5.0.13 (ref: manual).
           We have to not only seek but also
           a) init_master_info(), to seek back to hot relay log's start for later
           (for when we will come back to this hot log after re-processing the
@@ -3303,11 +1950,19 @@ pthread_handler_t handle_slave_io(void *arg)
   thd->proc_info = "Connecting to master";
   // we can get killed during safe_connect
   if (!safe_connect(thd, mysql, mi))
+  {
     sql_print_information("Slave I/O thread: connected to master '%s@%s:%d',\
   replication started in log '%s' at position %s", mi->user,
-                    mi->host, mi->port,
-                    IO_RPL_LOG_NAME,
-                    llstr(mi->master_log_pos,llbuff));
+			  mi->host, mi->port,
+			  IO_RPL_LOG_NAME,
+			  llstr(mi->master_log_pos,llbuff));
+  /*
+    Adding MAX_LOG_EVENT_HEADER_LEN to the max_packet_size on the I/O
+    thread, since a replication event can become this much larger than
+    the corresponding packet (query) sent from client to master.
+  */
+    mysql->net.max_packet_size= thd->net.max_packet_size+= MAX_LOG_EVENT_HEADER;
+  }
   else
   {
     sql_print_information("Slave I/O thread killed while connecting to master");
@@ -3632,7 +2287,7 @@ pthread_handler_t handle_slave_sql(void *arg)
     now.
     But the master timestamp is reset by RESET SLAVE & CHANGE MASTER.
   */
-  clear_slave_error(rli);
+  rli->clear_slave_error();
 
   //tell the I/O thread to take relay_log_space_limit into account from now on
   pthread_mutex_lock(&rli->log_space_lock);
@@ -4455,7 +3110,7 @@ static int connect_to_master(THD* thd, MYSQL* mysql, MASTER_INFO* mi,
 '%s@%s:%d': \
 Error: '%s'  errno: %d  retry-time: %d  retries: %lu",
                       (reconnect ? "reconnecting" : "connecting"),
-                      mi->user,mi->host,mi->port,
+                      mi->user, mi->host, mi->port,
                       mysql_error(mysql), last_errno,
                       mi->connect_retry,
                       master_retry_count);
@@ -4990,121 +3645,6 @@ end:
   DBUG_VOID_RETURN;
 }
 
-/*
-  Some system tables needed to be re-read by the MySQL server after it has
-  updated them; in statement-based replication, the GRANT and other commands
-  are sent verbatim to the slave which then reloads; in row-based replication,
-  changes to these tables are done through ordinary Rows binlog events, so
-  master must add some flag for the slave to know it has to reload the tables.
-*/
-struct st_reload_entry
-{
-  char const *table;
-  st_relay_log_info::enum_reload_flag flag;
-};
-
-/*
-  Sorted array of table names, please keep it sorted since we are
-  using bsearch() on it below.
- */
-static st_reload_entry s_mysql_tables[] =
-{
-  { "columns_priv", st_relay_log_info::RELOAD_GRANT_F },
-  { "db",           st_relay_log_info::RELOAD_ACCESS_F },
-  { "host",         st_relay_log_info::RELOAD_ACCESS_F },
-  { "procs_priv",   st_relay_log_info::RELOAD_GRANT_F },
-  { "tables_priv",  st_relay_log_info::RELOAD_GRANT_F },
-  { "user",         st_relay_log_info::RELOAD_ACCESS_F }
-};
-
-static const my_size_t s_mysql_tables_size =
-  sizeof(s_mysql_tables)/sizeof(*s_mysql_tables);
-
-static int reload_entry_compare(const void *lhs, const void *rhs)
-{
-  const char *lstr = static_cast<const char *>(lhs);
-  const char *rstr = static_cast<const st_reload_entry*>(rhs)->table;
-  DBUG_ENTER("reload_entry_compare");
-
-  DBUG_RETURN(strcmp(lstr, rstr));
-}
-
-void st_relay_log_info::touching_table(char const* db, char const* table,
-                                       ulong table_id)
-{
-  DBUG_ENTER("st_relay_log_info::touching_table");
-
-  if (strcmp(db,"mysql") == 0)
-  {
-#if defined(HAVE_BSEARCH) && defined(HAVE_SIZE_T)
-    void *const ptr= bsearch(table, s_mysql_tables,
-                             s_mysql_tables_size,
-                             sizeof(*s_mysql_tables), reload_entry_compare);
-    st_reload_entry const *const entry= static_cast<st_reload_entry*>(ptr);
-#else
-    /*
-      Fall back to full scan, there are few rows anyway and updating the
-      "mysql" database is rare.
-    */
-    st_reload_entry const *entry= s_mysql_tables;
-    for ( ; entry < s_mysql_tables + s_mysql_tables_size ; entry++)
-      if (reload_entry_compare(table, entry) == 0)
-        break;
-#endif
-    if (entry)
-      m_reload_flags|= entry->flag;
-  }
-  DBUG_VOID_RETURN;
-}
-
-void st_relay_log_info::transaction_end(THD* thd)
-{
-  DBUG_ENTER("st_relay_log_info::transaction_end");
-
-  if (m_reload_flags != RELOAD_NONE_F)
-  {
-    if (m_reload_flags & RELOAD_ACCESS_F)
-      acl_reload(thd);
-
-    if (m_reload_flags & RELOAD_GRANT_F)
-      grant_reload(thd);
-
-    m_reload_flags= RELOAD_NONE_F;
-  }
-  DBUG_VOID_RETURN;
-}
-
-#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
-void st_relay_log_info::cleanup_context(THD *thd, bool error)
-{
-  DBUG_ENTER("st_relay_log_info::cleanup_context");
-
-  DBUG_ASSERT(sql_thd == thd);
-  /*
-    1) Instances of Table_map_log_event, if ::exec_event() was called on them,
-    may have opened tables, which we cannot be sure have been closed (because
-    maybe the Rows_log_event have not been found or will not be, because slave
-    SQL thread is stopping, or relay log has a missing tail etc). So we close
-    all thread's tables. And so the table mappings have to be cancelled.
-    2) Rows_log_event::exec_event() may even have started statements or
-    transactions on them, which we need to rollback in case of error.
-    3) If finding a Format_description_log_event after a BEGIN, we also need
-    to rollback before continuing with the next events.
-    4) so we need this "context cleanup" function.
-  */
-  if (error)
-  {
-    ha_autocommit_or_rollback(thd, 1); // if a "statement transaction"
-    end_trans(thd, ROLLBACK); // if a "real transaction"
-  }
-  m_table_map.clear_tables();
-  close_thread_tables(thd);
-  clear_tables_to_lock();
-  unsafe_to_stop_at= 0;
-  DBUG_VOID_RETURN;
-}
-#endif
-
 
 #ifdef HAVE_EXPLICIT_TEMPLATE_INSTANTIATION
 template class I_List_iterator<i_string>;
diff --git a/sql/slave.h b/sql/slave.h
index e70b2e4b326..24ba09d78d3 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -24,6 +24,7 @@
 #include "rpl_filter.h"
 #include "rpl_tblmap.h"
 #include "rpl_rli.h"
+#include "rpl_mi.h"
 
 #define SLAVE_NET_TIMEOUT  3600
 
@@ -38,11 +39,11 @@
     I/O Thread - One of these threads is started for each master server.
                  They maintain a connection to their master server, read log
                  events from the master as they arrive, and queues them into
-                 a single, shared relay log file.  A MASTER_INFO struct
+                 a single, shared relay log file.  A MASTER_INFO 
                  represents each of these threads.
 
     SQL Thread - One of these threads is started and reads from the relay log
-                 file, executing each event.  A RELAY_LOG_INFO struct
+                 file, executing each event.  A RELAY_LOG_INFO 
                  represents this thread.
 
   Buffering in the relay log file makes it unnecessary to reread events from
@@ -95,7 +96,6 @@ extern my_string opt_relay_logname, opt_relaylog_index_name;
 extern my_bool opt_skip_slave_start, opt_reckless_slave;
 extern my_bool opt_log_slave_updates;
 extern ulonglong relay_log_space_limit;
-struct st_master_info;
 
 /*
   3 possible values for MASTER_INFO::slave_running and
@@ -114,110 +114,6 @@ struct st_master_info;
 
 static Log_event* next_event(RELAY_LOG_INFO* rli);
 
-/*****************************************************************************
-
-  Replication IO Thread
-
-  st_master_info contains:
-    - information about how to connect to a master
-    - current master log name
-    - current master log offset
-    - misc control variables
-
-  st_master_info is initialized once from the master.info file if such
-  exists. Otherwise, data members corresponding to master.info fields
-  are initialized with defaults specified by master-* options. The
-  initialization is done through init_master_info() call.
-
-  The format of master.info file:
-
-  log_name
-  log_pos
-  master_host
-  master_user
-  master_pass
-  master_port
-  master_connect_retry
-
-  To write out the contents of master.info file to disk ( needed every
-  time we read and queue data from the master ), a call to
-  flush_master_info() is required.
-
-  To clean up, call end_master_info()
-
-*****************************************************************************/
-
-typedef struct st_master_info
-{
-  /* the variables below are needed because we can change masters on the fly */
-  char master_log_name[FN_REFLEN];
-  char host[HOSTNAME_LENGTH+1];
-  char user[USERNAME_LENGTH+1];
-  char password[MAX_PASSWORD_LENGTH+1];
-  my_bool ssl; // enables use of SSL connection if true
-  char ssl_ca[FN_REFLEN], ssl_capath[FN_REFLEN], ssl_cert[FN_REFLEN];
-  char ssl_cipher[FN_REFLEN], ssl_key[FN_REFLEN];
-
-  my_off_t master_log_pos;
-  File fd; // we keep the file open, so we need to remember the file pointer
-  IO_CACHE file;
-
-  pthread_mutex_t data_lock,run_lock;
-  pthread_cond_t data_cond,start_cond,stop_cond;
-  THD *io_thd;
-  MYSQL* mysql;
-  uint32 file_id;				/* for 3.23 load data infile */
-  RELAY_LOG_INFO rli;
-  uint port;
-  uint connect_retry;
-#ifndef DBUG_OFF
-  int events_till_disconnect;
-#endif
-  bool inited;
-  volatile bool abort_slave;
-  volatile uint slave_running;
-  volatile ulong slave_run_id;
-  /*
-     The difference in seconds between the clock of the master and the clock of
-     the slave (second - first). It must be signed as it may be <0 or >0.
-     clock_diff_with_master is computed when the I/O thread starts; for this the
-     I/O thread does a SELECT UNIX_TIMESTAMP() on the master.
-     "how late the slave is compared to the master" is computed like this:
-     clock_of_slave - last_timestamp_executed_by_SQL_thread - clock_diff_with_master
-
-  */
-  long clock_diff_with_master;
-
-  st_master_info()
-    :ssl(0), fd(-1),  io_thd(0), inited(0),
-     abort_slave(0),slave_running(0), slave_run_id(0)
-  {
-    host[0] = 0; user[0] = 0; password[0] = 0;
-    ssl_ca[0]= 0; ssl_capath[0]= 0; ssl_cert[0]= 0;
-    ssl_cipher[0]= 0; ssl_key[0]= 0;
-
-    bzero((char*) &file, sizeof(file));
-    pthread_mutex_init(&run_lock, MY_MUTEX_INIT_FAST);
-    pthread_mutex_init(&data_lock, MY_MUTEX_INIT_FAST);
-    pthread_cond_init(&data_cond, NULL);
-    pthread_cond_init(&start_cond, NULL);
-    pthread_cond_init(&stop_cond, NULL);
-  }
-
-  ~st_master_info()
-  {
-    pthread_mutex_destroy(&run_lock);
-    pthread_mutex_destroy(&data_lock);
-    pthread_cond_destroy(&data_cond);
-    pthread_cond_destroy(&start_cond);
-    pthread_cond_destroy(&stop_cond);
-  }
-
-} MASTER_INFO;
-
-
-int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len);
-
 #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 :\
@@ -231,7 +127,6 @@ int queue_event(MASTER_INFO* mi,const char* buf,ulong event_len);
 
 int init_slave();
 void init_slave_skip_errors(const char* arg);
-int flush_master_info(MASTER_INFO* mi, bool flush_relay_log_cache);
 bool flush_relay_log_info(RELAY_LOG_INFO* rli);
 int register_slave_on_master(MYSQL* mysql);
 int terminate_slave_threads(MASTER_INFO* mi, int thread_mask,
@@ -276,14 +171,8 @@ void slave_print_msg(enum loglevel level, RELAY_LOG_INFO* rli,
   ATTRIBUTE_FORMAT(printf, 4, 5);
 
 void end_slave(); /* clean up */
-void init_master_info_with_options(MASTER_INFO* mi);
 void clear_until_condition(RELAY_LOG_INFO* rli);
 void clear_slave_error(RELAY_LOG_INFO* rli);
-int init_master_info(MASTER_INFO* mi, const char* master_info_fname,
-		     const char* slave_info_fname,
-		     bool abort_if_no_master_info_file,
-		     int thread_mask);
-void end_master_info(MASTER_INFO* mi);
 void end_relay_log_info(RELAY_LOG_INFO* rli);
 void lock_slave_threads(MASTER_INFO* mi);
 void unlock_slave_threads(MASTER_INFO* mi);
diff --git a/sql/sql_binlog.cc b/sql/sql_binlog.cc
index 23ca5330053..732981b58f3 100644
--- a/sql/sql_binlog.cc
+++ b/sql/sql_binlog.cc
@@ -79,9 +79,15 @@ void mysql_client_binlog_statement(THD* thd)
     char const *endptr= 0;
     int bytes_decoded= base64_decode(strptr, coded_len, buf, &endptr);
 
+#ifndef HAVE_purify
+      /*
+        This debug printout should not be used for valgrind builds
+        since it will read from unassigned memory.
+      */
     DBUG_PRINT("info",
                ("bytes_decoded=%d; strptr=0x%lu; endptr=0x%lu ('%c':%d)",
                 bytes_decoded, strptr, endptr, *endptr, *endptr));
+#endif
 
     if (bytes_decoded < 0)
     {
@@ -147,8 +153,14 @@ void mysql_client_binlog_statement(THD* thd)
       DBUG_PRINT("info",("ev->get_type_code()=%d", ev->get_type_code()));
       DBUG_PRINT("info",("bufptr+EVENT_TYPE_OFFSET=0x%lx",
                          bufptr+EVENT_TYPE_OFFSET));
+#ifndef HAVE_purify
+      /*
+        This debug printout should not be used for valgrind builds
+        since it will read from unassigned memory.
+      */
       DBUG_PRINT("info", ("bytes_decoded=%d; bufptr=0x%lx; buf[EVENT_LEN_OFFSET]=%u",
                           bytes_decoded, bufptr, uint4korr(bufptr+EVENT_LEN_OFFSET)));
+#endif
       ev->thd= thd;
       if (int err= ev->exec_event(thd->rli_fake))
       {
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 776ba4dabea..20acb28bc77 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -2552,6 +2552,128 @@ my_size_t THD::pack_row(TABLE *table, MY_BITMAP const* cols, byte *row_data,
 }
 
 
+namespace {
+  /**
+     Class to handle temporary allocation of memory for row data.
+
+     The responsibilities of the class is to provide memory for
+     packing one or two rows of packed data (depending on what
+     constructor is called).
+
+     In order to make the allocation more efficient for "simple" rows,
+     i.e., rows that do not contain any blobs, a pointer to the
+     allocated memory is of memory is stored in the table structure
+     for simple rows.  If memory for a table containing a blob field
+     is requested, only memory for that is allocated, and subsequently
+     released when the object is destroyed.
+
+   */
+  class Row_data_memory {
+  public:
+    /**
+      Build an object to keep track of a block-local piece of memory
+      for storing a row of data.
+
+      @param table
+      Table where the pre-allocated memory is stored.
+
+      @param length
+      Length of data that is needed, if the record contain blobs.
+     */
+    Row_data_memory(TABLE *table, my_size_t const len1)
+      : m_memory(0)
+    {
+#ifndef DBUG_OFF
+      m_alloc_checked= false;
+#endif
+      allocate_memory(table, len1);
+      m_ptr[0]= has_memory() ? m_memory : 0;
+      m_ptr[1]= 0;
+    }
+
+    Row_data_memory(TABLE *table, my_size_t const len1, my_size_t const len2)
+      : m_memory(0)
+    {
+#ifndef DBUG_OFF
+      m_alloc_checked= false;
+#endif
+      allocate_memory(table, len1 + len2);
+      m_ptr[0]= has_memory() ? m_memory        : 0;
+      m_ptr[1]= has_memory() ? m_memory + len1 : 0;
+    }
+
+    ~Row_data_memory()
+    {
+      if (m_memory != 0 && m_release_memory_on_destruction)
+        my_free((gptr) m_memory, MYF(MY_WME));
+    }
+
+    /**
+       Is there memory allocated?
+
+       @retval true There is memory allocated
+       @retval false Memory allocation failed
+     */
+    bool has_memory() const {
+#ifndef DBUG_OFF
+      m_alloc_checked= true;
+#endif
+      return m_memory != 0;
+    }
+
+    byte *slot(int const s)
+    {
+      DBUG_ASSERT(0 <= s && s < sizeof(m_ptr)/sizeof(*m_ptr));
+      DBUG_ASSERT(m_ptr[s] != 0);
+      DBUG_ASSERT(m_alloc_checked == true);
+      return m_ptr[s];
+    }
+
+  private:
+    void allocate_memory(TABLE *const table, my_size_t const total_length)
+    {
+      if (table->s->blob_fields == 0)
+      {
+        /*
+          The maximum length of a packed record is less than this
+          length. We use this value instead of the supplied length
+          when allocating memory for records, since we don't know how
+          the memory will be used in future allocations.
+
+          Since table->s->reclength is for unpacked records, we have
+          to add two bytes for each field, which can potentially be
+          added to hold the length of a packed field.
+        */
+        my_size_t const maxlen= table->s->reclength + 2 * table->s->fields;
+
+        /*
+          Allocate memory for two records if memory hasn't been
+          allocated. We allocate memory for two records so that it can
+          be used when processing update rows as well.
+        */
+        if (table->write_row_record == 0)
+          table->write_row_record=
+            (byte *) alloc_root(&table->mem_root, 2 * maxlen);
+        m_memory= table->write_row_record;
+        m_release_memory_on_destruction= false;
+      }
+      else
+      {
+        m_memory= (byte *) my_malloc(total_length, MYF(MY_WME));
+        m_release_memory_on_destruction= true;
+      }
+    }
+
+#ifndef DBUG_OFF
+    mutable bool m_alloc_checked;
+#endif
+    bool m_release_memory_on_destruction;
+    byte *m_memory;
+    byte *m_ptr[2];
+  };
+}
+
+
 int THD::binlog_write_row(TABLE* table, bool is_trans, 
                           MY_BITMAP const* cols, my_size_t colcnt, 
                           byte const *record) 
@@ -2562,40 +2684,25 @@ int THD::binlog_write_row(TABLE* table, bool is_trans,
     Pack records into format for transfer. We are allocating more
     memory than needed, but that doesn't matter.
   */
-  bool error= 0;
-  byte *row_data= table->write_row_record;
-  my_size_t const max_len= max_row_length(table, record);
-  my_size_t len;
-  Rows_log_event *ev;
-  
-  /* Allocate room for a row (if needed) */
-  if (!row_data)
-  {
-    if (!table->s->blob_fields)
-    {
-      /* multiply max_len by 2 so it can be used for update_row as well */
-      table->write_row_record= (byte *) alloc_root(&table->mem_root,
-                                                   2*max_len);
-      if (!table->write_row_record)
-        return HA_ERR_OUT_OF_MEM;
-      row_data= table->write_row_record;
-    }
-    else if (unlikely(!(row_data= (byte *) my_malloc(max_len, MYF(MY_WME)))))
-      return HA_ERR_OUT_OF_MEM;
-  }
-  len= pack_row(table, cols, row_data, record);
+  int error= 0;
 
-  ev= binlog_prepare_pending_rows_event(table, server_id, cols, colcnt,
-                                        len, is_trans,
-                                        static_cast<Write_rows_log_event*>(0));
+  Row_data_memory memory(table, max_row_length(table, record));
+  if (!memory.has_memory())
+    return HA_ERR_OUT_OF_MEM;
 
-  /* add_row_data copies row_data to internal buffer */
-  error= likely(ev != 0) ? ev->add_row_data(row_data,len) : HA_ERR_OUT_OF_MEM ;
+  byte *row_data= memory.slot(0);
 
-  if (table->write_row_record == 0)
-    my_free((gptr) row_data, MYF(MY_WME));
+  my_size_t const len= pack_row(table, cols, row_data, record);
 
-  return error;
+  Rows_log_event* const ev=
+    binlog_prepare_pending_rows_event(table, server_id, cols, colcnt,
+                                      len, is_trans,
+                                      static_cast<Write_rows_log_event*>(0));
+
+  if (unlikely(ev == 0))
+    return HA_ERR_OUT_OF_MEM;
+
+  return ev->add_row_data(row_data, len);
 }
 
 int THD::binlog_update_row(TABLE* table, bool is_trans,
@@ -2605,53 +2712,44 @@ int THD::binlog_update_row(TABLE* table, bool is_trans,
 { 
   DBUG_ASSERT(current_stmt_binlog_row_based && mysql_bin_log.is_open());
 
-  bool error= 0;
+  int error= 0;
   my_size_t const before_maxlen = max_row_length(table, before_record);
   my_size_t const after_maxlen  = max_row_length(table, after_record);
 
-  byte *row_data= table->write_row_record;
-  byte *before_row, *after_row;
-  if (row_data != 0)
-  {
-    before_row= row_data;
-    after_row= before_row + before_maxlen;
-  }
-  else
-  {
-    if (unlikely(!(row_data= (byte*)my_multi_malloc(MYF(MY_WME),
-                                             &before_row, before_maxlen,
-                                             &after_row, after_maxlen,
-                                             NULL))))
-      return HA_ERR_OUT_OF_MEM;
-  }
+  Row_data_memory row_data(table, before_maxlen, after_maxlen);
+  if (!row_data.has_memory())
+    return HA_ERR_OUT_OF_MEM;
+
+  byte *before_row= row_data.slot(0);
+  byte *after_row= row_data.slot(1);
 
   my_size_t const before_size= pack_row(table, cols, before_row, 
                                         before_record);
   my_size_t const after_size= pack_row(table, cols, after_row, 
                                        after_record);
-  
+
+  /*
+    Don't print debug messages when running valgrind since they can
+    trigger false warnings.
+   */
+#ifndef HAVE_purify
   DBUG_DUMP("before_record", (const char *)before_record, table->s->reclength);
   DBUG_DUMP("after_record", (const char *)after_record, table->s->reclength);
   DBUG_DUMP("before_row", (const char *)before_row, before_size);
   DBUG_DUMP("after_row", (const char *)after_row, after_size);
+#endif
 
   Rows_log_event* const ev=
     binlog_prepare_pending_rows_event(table, server_id, cols, colcnt,
 				      before_size + after_size, is_trans,
 				      static_cast<Update_rows_log_event*>(0));
 
-  error=
-    unlikely(!ev) ||
+  if (unlikely(ev == 0))
+    return HA_ERR_OUT_OF_MEM;
+
+  return
     ev->add_row_data(before_row, before_size) ||
     ev->add_row_data(after_row, after_size);
-
-  if (!table->write_row_record)
-  {
-    /* add_row_data copies row_data to internal buffer */
-    my_free((gptr)row_data, MYF(MY_WME));
-  }
-  
-  return error;
 }
 
 int THD::binlog_delete_row(TABLE* table, bool is_trans, 
@@ -2664,11 +2762,14 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
      Pack records into format for transfer. We are allocating more
      memory than needed, but that doesn't matter.
   */
-  bool error= 0;
-  my_size_t const max_len= max_row_length(table, record);
-  byte *row_data= table->write_row_record;
-  if (!row_data && unlikely(!(row_data= (byte*)my_malloc(max_len, MYF(MY_WME)))))
+  int error= 0;
+
+  Row_data_memory memory(table, max_row_length(table, record));
+  if (unlikely(!memory.has_memory()))
     return HA_ERR_OUT_OF_MEM;
+
+  byte *row_data= memory.slot(0);
+
   my_size_t const len= pack_row(table, cols, row_data, record);
 
   Rows_log_event* const ev=
@@ -2676,13 +2777,10 @@ int THD::binlog_delete_row(TABLE* table, bool is_trans,
 				      len, is_trans,
 				      static_cast<Delete_rows_log_event*>(0));
 
-  error= (unlikely(!ev)) || ev->add_row_data(row_data, len);
-
-  /* add_row_data copies row_data */
-  if (table->write_row_record == 0)
-    my_free((gptr)row_data, MYF(MY_WME));
+  if (unlikely(ev == 0))
+    return HA_ERR_OUT_OF_MEM;
 
-  return error;
+  return ev->add_row_data(row_data, len);
 }
 
 
@@ -2776,6 +2874,12 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype,
 #endif /*HAVE_ROW_BASED_REPLICATION*/
 
   switch (qtype) {
+  case THD::ROW_QUERY_TYPE:
+#ifdef HAVE_ROW_BASED_REPLICATION
+    if (current_stmt_binlog_row_based)
+      DBUG_RETURN(0);
+#endif
+    /* Otherwise, we fall through */
   case THD::MYSQL_QUERY_TYPE:
     /*
       Using this query type is a conveniece hack, since we have been
@@ -2785,12 +2889,6 @@ int THD::binlog_query(THD::enum_binlog_query_type qtype,
       Make sure to change in check_table_binlog_row_based() according
       to how you treat this.
     */
-  case THD::ROW_QUERY_TYPE:
-#ifdef HAVE_ROW_BASED_REPLICATION
-    if (current_stmt_binlog_row_based)
-      DBUG_RETURN(0);
-#endif
-    /* Otherwise, we fall through */
   case THD::STMT_QUERY_TYPE:
     /*
       The MYSQL_LOG::write() function will set the STMT_END_F flag and
diff --git a/sql/sql_parse.cc b/sql/sql_parse.cc
index 2a45520af81..5b61a5fb48a 100644
--- a/sql/sql_parse.cc
+++ b/sql/sql_parse.cc
@@ -2550,7 +2550,23 @@ mysql_execute_command(THD *thd)
     {
       /* we warn the slave SQL thread */
       my_message(ER_SLAVE_IGNORED_TABLE, ER(ER_SLAVE_IGNORED_TABLE), MYF(0));
-      reset_one_shot_variables(thd);
+      if (thd->one_shot_set)
+      {
+        /*
+          It's ok to check thd->one_shot_set here:
+
+          The charsets in a MySQL 5.0 slave can change by both a binlogged
+          SET ONE_SHOT statement and the event-internal charset setting, 
+          and these two ways to change charsets do not seems to work
+          together.
+
+          At least there seems to be problems in the rli cache for
+          charsets if we are using ONE_SHOT.  Note that this is normally no
+          problem because either the >= 5.0 slave reads a 4.1 binlog (with
+          ONE_SHOT) *or* or 5.0 binlog (without ONE_SHOT) but never both."
+        */
+        reset_one_shot_variables(thd);
+      }
       DBUG_RETURN(0);
     }
   }
@@ -6317,7 +6333,7 @@ TABLE_LIST *st_select_lex::add_table_to_list(THD *thd,
 
   ptr->alias= alias_str;
   if (lower_case_table_names && table->table.length)
-    my_casedn_str(files_charset_info, table->table.str);
+    table->table.length= my_casedn_str(files_charset_info, table->table.str);
   ptr->table_name=table->table.str;
   ptr->table_name_length=table->table.length;
   ptr->lock_type=   lock_type;
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 52489087b02..7ffbb649cb8 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -431,6 +431,12 @@ impossible position";
     goto err;
   }
   packet->set("\0", 1, &my_charset_bin);
+  /*
+    Adding MAX_LOG_EVENT_HEADER_LEN, since a binlog event can become
+    this larger than the corresponding packet (query) sent 
+    from client to master.
+  */
+  thd->variables.max_allowed_packet+= MAX_LOG_EVENT_HEADER;
 
   /*
     We can set log_lock now, it does not move (it's a member of
@@ -805,7 +811,7 @@ int start_slave(THD* thd , MASTER_INFO* mi,  bool net_report)
                   sizeof(mi->rli.until_log_name)-1);
         }
         else
-          clear_until_condition(&mi->rli);
+          mi->rli.clear_until_condition();
 
         if (mi->rli.until_condition != RELAY_LOG_INFO::UNTIL_NONE)
         {
@@ -978,8 +984,8 @@ int reset_slave(THD *thd, MASTER_INFO* mi)
      Reset errors (the idea is that we forget about the
      old master).
   */
-  clear_slave_error(&mi->rli);
-  clear_until_condition(&mi->rli);
+  mi->rli.clear_slave_error();
+  mi->rli.clear_until_condition();
 
   // close master_info_file, relay_log_info_file, set mi->inited=rli->inited=0
   end_master_info(mi);
@@ -1244,8 +1250,8 @@ bool change_master(THD* thd, MASTER_INFO* mi)
   pthread_mutex_lock(&mi->rli.data_lock);
   mi->rli.abort_pos_wait++; /* for MASTER_POS_WAIT() to abort */
   /* Clear the errors, for a clean start */
-  clear_slave_error(&mi->rli);
-  clear_until_condition(&mi->rli);
+  mi->rli.clear_slave_error();
+  mi->rli.clear_until_condition();
   /*
     If we don't write new coordinates to disk now, then old will remain in
     relay-log.info until START SLAVE is issued; but if mysqld is shutdown
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index d40d9f16bb5..6758f9d68d0 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -8723,6 +8723,7 @@ static Field *create_tmp_field_from_item(THD *thd, Item *item, TABLE *table,
                                      item->collation.collation);
     else
       new_field= item->make_string_field(table);
+    new_field->set_derivation(item->collation.derivation);
     break;
   case DECIMAL_RESULT:
     new_field= new Field_new_decimal(item->max_length, maybe_null, item->name,
@@ -8908,7 +8909,9 @@ Field *create_tmp_field(THD *thd, TABLE *table,Item *item, Item::Type type,
                                       (make_copy_field ? 0 : copy_func),
                                        modify_item, convert_blob_length);
   case Item::TYPE_HOLDER:  
-    return ((Item_type_holder *)item)->make_field_by_type(table);
+    result= ((Item_type_holder *)item)->make_field_by_type(table);
+    result->set_derivation(item->collation.derivation);
+    return result;
   default:					// Dosen't have to be stored
     return 0;
   }
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index f6e39fb7913..dc8946df876 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1058,6 +1058,9 @@ int store_create_info(THD *thd, TABLE_LIST *table_list, String *packet,
     packet->append(STRING_WITH_LEN("CREATE TEMPORARY TABLE "));
   else
     packet->append(STRING_WITH_LEN("CREATE TABLE "));
+  if (create_info_arg &&
+      (create_info_arg->options & HA_LEX_CREATE_IF_NOT_EXISTS))
+    packet->append(STRING_WITH_LEN("IF NOT EXISTS "));
   if (table_list->schema_table)
     alias= table_list->schema_table->table_name;
   else
diff --git a/sql/sql_string.cc b/sql/sql_string.cc
index 6e4d3f2ed0a..29b53560067 100644
--- a/sql/sql_string.cc
+++ b/sql/sql_string.cc
@@ -844,6 +844,162 @@ outp:
 }
 
 
+/*
+  copy a string,
+  with optional character set conversion,
+  with optional left padding (for binary -> UCS2 conversion)
+  
+  SYNOPSIS
+    well_formed_copy_nhars()
+    to			     Store result here
+    to_length                Maxinum length of "to" string
+    to_cs		     Character set of "to" string
+    from		     Copy from here
+    from_length		     Length of from string
+    from_cs		     From character set
+    nchars                   Copy not more that nchars characters
+    well_formed_error_pos    Return position when "from" is not well formed
+                             or NULL otherwise.
+    cannot_convert_error_pos Return position where a not convertable
+                             character met, or NULL otherwise.
+    from_end_pos             Return position where scanning of "from"
+                             string stopped.
+  NOTES
+
+  RETURN
+    length of bytes copied to 'to'
+*/
+
+
+uint32
+well_formed_copy_nchars(CHARSET_INFO *to_cs,
+                        char *to, uint to_length,
+                        CHARSET_INFO *from_cs,
+                        const char *from, uint from_length,
+                        uint nchars,
+                        const char **well_formed_error_pos,
+                        const char **cannot_convert_error_pos,
+                        const char **from_end_pos)
+{
+  uint res;
+
+  if ((to_cs == &my_charset_bin) || 
+      (from_cs == &my_charset_bin) ||
+      (to_cs == from_cs) ||
+      my_charset_same(from_cs, to_cs))
+  {
+    if (to_length < to_cs->mbminlen || !nchars)
+    {
+      *from_end_pos= from;
+      *cannot_convert_error_pos= NULL;
+      *well_formed_error_pos= NULL;
+      return 0;
+    }
+
+    if (to_cs == &my_charset_bin)
+    {
+      res= min(min(nchars, to_length), from_length);
+      memmove(to, from, res);
+      *from_end_pos= from + res;
+      *well_formed_error_pos= NULL;
+      *cannot_convert_error_pos= NULL;
+    }
+    else
+    {
+      int well_formed_error;
+      uint from_offset;
+
+      if ((from_offset= (from_length % to_cs->mbminlen)) &&
+          (from_cs == &my_charset_bin))
+      {
+        /*
+          Copying from BINARY to UCS2 needs to prepend zeros sometimes:
+          INSERT INTO t1 (ucs2_column) VALUES (0x01);
+          0x01 -> 0x0001
+        */
+        uint pad_length= to_cs->mbminlen - from_offset;
+        bzero(to, pad_length);
+        memmove(to + pad_length, from, from_offset);
+        nchars--;
+        from+= from_offset;
+        from_length-= from_offset;
+        to+= to_cs->mbminlen;
+        to_length-= to_cs->mbminlen;
+      }
+
+      set_if_smaller(from_length, to_length);
+      res= to_cs->cset->well_formed_len(to_cs, from, from + from_length,
+                                        nchars, &well_formed_error);
+      memmove(to, from, res);
+      *from_end_pos= from + res;
+      *well_formed_error_pos= well_formed_error ? from + res : NULL;
+      *cannot_convert_error_pos= NULL;
+      if (from_offset)
+        res+= to_cs->mbminlen;
+    }
+  }
+  else
+  {
+    int cnvres;
+    my_wc_t wc;
+    int (*mb_wc)(struct charset_info_st *, my_wc_t *,
+                 const uchar *, const uchar *)= from_cs->cset->mb_wc;
+    int (*wc_mb)(struct charset_info_st *, my_wc_t,
+                 uchar *s, uchar *e)= to_cs->cset->wc_mb;
+    const uchar *from_end= (const uchar*) from + from_length;
+    uchar *to_end= (uchar*) to + to_length;
+    char *to_start= to;
+    *well_formed_error_pos= NULL;
+    *cannot_convert_error_pos= NULL;
+
+    for ( ; nchars; nchars--)
+    {
+      const char *from_prev= from;
+      if ((cnvres= (*mb_wc)(from_cs, &wc, (uchar*) from, from_end)) > 0)
+        from+= cnvres;
+      else if (cnvres == MY_CS_ILSEQ)
+      {
+        if (!*well_formed_error_pos)
+          *well_formed_error_pos= from;
+        from++;
+        wc= '?';
+      }
+      else if (cnvres > MY_CS_TOOSMALL)
+      {
+        /*
+          A correct multibyte sequence detected
+          But it doesn't have Unicode mapping.
+        */
+        if (!*cannot_convert_error_pos)
+          *cannot_convert_error_pos= from;
+        from+= (-cnvres);
+        wc= '?';
+      }
+      else
+        break;  // Not enough characters
+
+outp:
+      if ((cnvres= (*wc_mb)(to_cs, wc, (uchar*) to, to_end)) > 0)
+        to+= cnvres;
+      else if (cnvres == MY_CS_ILUNI && wc != '?')
+      {
+        if (!*cannot_convert_error_pos)
+          *cannot_convert_error_pos= from_prev;
+        wc= '?';
+        goto outp;
+      }
+      else
+        break;
+    }
+    *from_end_pos= from;
+    res= to - to_start;
+  }
+  return (uint32) res;
+}
+
+
+
+
 void String::print(String *str)
 {
   char *st= (char*)Ptr, *end= st+str_length;
diff --git a/sql/sql_string.h b/sql/sql_string.h
index b1d417be2c2..a72b24ae9d0 100644
--- a/sql/sql_string.h
+++ b/sql/sql_string.h
@@ -30,6 +30,14 @@ String *copy_if_not_alloced(String *a,String *b,uint32 arg_length);
 uint32 copy_and_convert(char *to, uint32 to_length, CHARSET_INFO *to_cs,
 			const char *from, uint32 from_length,
 			CHARSET_INFO *from_cs, uint *errors);
+uint32 well_formed_copy_nchars(CHARSET_INFO *to_cs,
+                               char *to, uint to_length,
+                               CHARSET_INFO *from_cs,
+                               const char *from, uint from_length,
+                               uint nchars,
+                               const char **well_formed_error_pos,
+                               const char **cannot_convert_error_pos,
+                               const char **from_end_pos);
 
 class String
 {
diff --git a/sql/sql_view.cc b/sql/sql_view.cc
index 5bf67af9271..98226c1651b 100644
--- a/sql/sql_view.cc
+++ b/sql/sql_view.cc
@@ -1695,24 +1695,23 @@ mysql_rename_view(THD *thd,
                   const char *new_name,
                   TABLE_LIST *view)
 {
-  LEX_STRING pathstr, file;
+  LEX_STRING pathstr;
   File_parser *parser;
-  char view_path[FN_REFLEN];
+  char path_buff[FN_REFLEN];
   bool error= TRUE;
   DBUG_ENTER("mysql_rename_view");
 
-  strxnmov(view_path, FN_REFLEN-1, mysql_data_home, "/", view->db, "/",
-           view->table_name, reg_ext, NullS);
-  (void) unpack_filename(view_path, view_path);
-
-  pathstr.str= (char *)view_path;
-  pathstr.length= strlen(view_path);
+  pathstr.str= (char *) path_buff;
+  pathstr.length= build_table_filename(path_buff, sizeof(path_buff) - 1,
+                                       view->db, view->table_name,
+                                       reg_ext, 0);
 
   if ((parser= sql_parse_prepare(&pathstr, thd->mem_root, 1)) && 
        is_equal(&view_type, parser->type()))
   {
     TABLE_LIST view_def;
-    char dir_buff[FN_REFLEN], file_buff[FN_REFLEN];
+    char dir_buff[FN_REFLEN];
+    LEX_STRING dir, file;
 
     /*
       To be PS-friendly we should either to restore state of
@@ -1735,18 +1734,18 @@ mysql_rename_view(THD *thd,
                               view_def.revision - 1, num_view_backups))
       goto err;
 
-    strxnmov(dir_buff, FN_REFLEN-1, mysql_data_home, "/", view->db, "/",
-             NullS);
-    (void) unpack_filename(dir_buff, dir_buff);
+    dir.str= dir_buff;
+    dir.length= build_table_filename(dir_buff, sizeof(dir_buff) - 1,
+                                     view->db, "", "", 0);
 
-    pathstr.str=    (char*)dir_buff;
-    pathstr.length= strlen(dir_buff);
+    pathstr.str= path_buff;
+    pathstr.length= build_table_filename(path_buff, sizeof(path_buff) - 1,
+                                      view->db, new_name, reg_ext, 0);
 
-    file.str= file_buff;
-    file.length= (strxnmov(file_buff, FN_REFLEN, new_name, reg_ext, NullS) 
-                  - file_buff);
+    file.str= pathstr.str + dir.length;
+    file.length= pathstr.length - dir.length;
 
-    if (sql_create_definition_file(&pathstr, &file, view_file_type,
+    if (sql_create_definition_file(&dir, &file, view_file_type,
                                    (gptr)&view_def, view_parameters,
                                    num_view_backups)) 
     {