Merge mysql_cab_desk.:C:/source/c++/mysql-5.1-new-rpl

into  mysql_cab_desk.:C:/source/c++/mysql-5.1_BUG_22086

6 files changed, 389 insertions, 133 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 37cdf452652..b1b89d16bbb 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1360,6 +1360,27 @@ bool Field::send_binary(Protocol *protocol)
 }
 
 
+/**
+   Check to see if field size is compatible with destination.
+
+   This method is used in row-based replication to verify that the slave's
+   field size is less than or equal to the master's field size. The 
+   encoded field metadata (from the master or source) is decoded and compared
+   to the size of this field (the slave or destination). 
+
+   @param   field_metadata   Encoded size in field metadata
+
+   @retval 0 if this field's size is < the source field's size
+   @retval 1 if this field's size is >= the source field's size
+*/
+int Field::compatible_field_size(uint field_metadata)
+{
+  uint const source_size= pack_length_from_metadata(field_metadata);
+  uint const destination_size= row_pack_length();
+  return (source_size <= destination_size);
+}
+
+
 int Field::store(const char *to, uint length, CHARSET_INFO *cs,
                  enum_check_fields check_level)
 {
@@ -2690,6 +2711,76 @@ void Field_new_decimal::sql_type(String &str) const
 }
 
 
+/**
+   Save the field metadata for new decimal fields.
+
+   Saves the precision in the first byte and decimals() in the second
+   byte of the field metadata array at index of *metadata_ptr and 
+   *(metadata_ptr + 1).
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_new_decimal::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= precision;
+  *(metadata_ptr + 1)= decimals();
+  return 2;
+}
+
+
+/**
+   Returns the number of bytes field uses in row-based replication 
+   row packed size.
+
+   This method is used in row-based replication to determine the number
+   of bytes that the field consumes in the row record format. This is
+   used to skip fields in the master that do not exist on the slave.
+
+   @param   field_metadata   Encoded size in field metadata
+
+   @returns The size of the field based on the field metadata.
+*/
+uint Field_new_decimal::pack_length_from_metadata(uint field_metadata)
+{
+  uint const source_precision= (field_metadata >> 8U) & 0x00ff;
+  uint const source_decimal= field_metadata & 0x00ff; 
+  uint const source_size= my_decimal_get_binary_size(source_precision, 
+                                                     source_decimal);
+  return (source_size);
+}
+
+
+/**
+   Check to see if field size is compatible with destination.
+
+   This method is used in row-based replication to verify that the slave's
+   field size is less than or equal to the master's field size. The 
+   encoded field metadata (from the master or source) is decoded and compared
+   to the size of this field (the slave or destination). 
+
+   @param   field_metadata   Encoded size in field metadata
+
+   @retval 0 if this field's size is < the source field's size
+   @retval 1 if this field's size is >= the source field's size
+*/
+int Field_new_decimal::compatible_field_size(uint field_metadata)
+{
+  int compatible= 0;
+  uint const source_precision= (field_metadata >> 8U) & 0x00ff;
+  uint const source_decimal= field_metadata & 0x00ff; 
+  uint const source_size= my_decimal_get_binary_size(source_precision, 
+                                                     source_decimal);
+  uint const destination_size= row_pack_length();
+  compatible= (source_size <= destination_size);
+  if (compatible)
+    compatible= (source_precision <= precision) &&
+                (source_decimal <= decimals());
+  return (compatible);
+}
+
+
 uint Field_new_decimal::is_equal(Create_field *new_field)
 {
   return ((new_field->sql_type == real_type()) &&
@@ -2724,7 +2815,9 @@ const uchar *Field_new_decimal::unpack(uchar* to,
   uint from_pack_len= my_decimal_get_binary_size(from_precision, from_decimal);
   uint len= (param_data && (from_pack_len < length)) ?
             from_pack_len : length;
-  if (from_pack_len && (from_pack_len < length))
+  if ((from_pack_len && (from_pack_len < length)) ||
+      (from_precision < precision) ||
+      (from_decimal < decimals()))
   {
     /*
       If the master's data is smaller than the slave, we need to convert
@@ -4087,6 +4180,22 @@ bool Field_float::send_binary(Protocol *protocol)
 }
 
 
+/**
+   Save the field metadata for float fields.
+
+   Saves the pack length in the first byte.
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_float::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= pack_length();
+  return 1;
+}
+
+
 void Field_float::sql_type(String &res) const
 {
   if (dec == NOT_FIXED_DEC)
@@ -4404,6 +4513,23 @@ void Field_double::sort_string(uchar *to,uint length __attribute__((unused)))
 }
 
 
+/**
+   Save the field metadata for double fields.
+
+   Saves the pack length in the first byte of the field metadata array
+   at index of *metadata_ptr.
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_double::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= pack_length();
+  return 1;
+}
+
+
 void Field_double::sql_type(String &res) const
 {
   CHARSET_INFO *cs=res.charset();
@@ -6445,6 +6571,25 @@ const uchar *Field_string::unpack(uchar *to, const uchar *from)
 }
 
 
+/**
+   Save the field metadata for string fields.
+
+   Saves the real type in the first byte and the field length in the 
+   second byte of the field metadata array at index of *metadata_ptr and
+   *(metadata_ptr + 1).
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_string::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= real_type();
+  *(metadata_ptr + 1)= field_length;
+  return 2;
+}
+
+
 /*
   Compare two packed keys
 
@@ -6597,6 +6742,24 @@ Field *Field_string::new_field(MEM_ROOT *root, struct st_table *new_table,
 
 const uint Field_varstring::MAX_SIZE= UINT_MAX16;
 
+/**
+   Save the field metadata for varstring fields.
+
+   Saves the field length in the first byte. Note: may consume
+   2 bytes. Caller must ensure second byte is contiguous with
+   first byte (e.g. array index 0,1).
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_varstring::do_save_field_metadata(uchar *metadata_ptr)
+{
+  char *ptr= (char *)metadata_ptr;
+  int2store(ptr, field_length);
+  return 2;
+}
+
 int Field_varstring::store(const char *from,uint length,CHARSET_INFO *cs)
 {
   ASSERT_COLUMN_MARKED_FOR_WRITE;
@@ -7563,6 +7726,23 @@ int Field_blob::key_cmp(const uchar *a,const uchar *b)
 }
 
 
+/**
+   Save the field metadata for blob fields.
+
+   Saves the pack length in the first byte of the field metadata array
+   at index of *metadata_ptr.
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_blob::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= pack_length_no_ptr();
+  return 1;
+}
+
+
 uint32 Field_blob::sort_length() const
 {
   return (uint32) (current_thd->variables.max_sort_length + 
@@ -8151,6 +8331,25 @@ longlong Field_enum::val_int(void)
 }
 
 
+/**
+   Save the field metadata for enum fields.
+
+   Saves the real type in the first byte and the pack length in the 
+   second byte of the field metadata array at index of *metadata_ptr and
+   *(metadata_ptr + 1).
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_enum::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= real_type();
+  *(metadata_ptr + 1)= pack_length();
+  return 2;
+}
+
+
 String *Field_enum::val_str(String *val_buffer __attribute__((unused)),
 			    String *val_ptr)
 {
@@ -8687,6 +8886,77 @@ uint Field_bit::get_key_image(uchar *buff, uint length, imagetype type_arg)
 }
 
 
+/**
+   Save the field metadata for bit fields.
+
+   Saves the bit length in the first byte and bytes in record in the
+   second byte of the field metadata array at index of *metadata_ptr and
+   *(metadata_ptr + 1).
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns number of bytes written to metadata_ptr
+*/
+int Field_bit::do_save_field_metadata(uchar *metadata_ptr)
+{
+  *metadata_ptr= bit_len;
+  *(metadata_ptr + 1)= bytes_in_rec;
+  return 2;
+}
+
+
+/**
+   Returns the number of bytes field uses in row-based replication 
+   row packed size.
+
+   This method is used in row-based replication to determine the number
+   of bytes that the field consumes in the row record format. This is
+   used to skip fields in the master that do not exist on the slave.
+
+   @param   field_metadata   Encoded size in field metadata
+
+   @returns The size of the field based on the field metadata.
+*/
+uint Field_bit::pack_length_from_metadata(uint field_metadata)
+{
+  uint const from_len= (field_metadata >> 8U) & 0x00ff;
+  uint const from_bit_len= field_metadata & 0x00ff;
+  uint const source_size= from_len + ((from_bit_len > 0) ? 1 : 0);
+  return (source_size);
+}
+
+
+/**
+   Check to see if field size is compatible with destination.
+
+   This method is used in row-based replication to verify that the slave's
+   field size is less than or equal to the master's field size. The 
+   encoded field metadata (from the master or source) is decoded and compared
+   to the size of this field (the slave or destination). 
+
+   @param   field_metadata   Encoded size in field metadata
+
+   @retval 0 if this field's size is < the source field's size
+   @retval 1 if this field's size is >= the source field's size
+*/
+int Field_bit::compatible_field_size(uint field_metadata)
+{
+  int compatible= 0;
+  uint const source_size= pack_length_from_metadata(field_metadata);
+  uint const destination_size= row_pack_length();
+  uint const from_bit_len= field_metadata & 0x00ff;
+  uint const from_len= (field_metadata >> 8U) & 0x00ff;
+  if ((bit_len == 0) || (from_bit_len == 0))
+    compatible= (source_size <= destination_size);
+  else if (from_bit_len > bit_len)
+    compatible= (from_len < bytes_in_rec);
+  else
+    compatible= ((from_bit_len <= bit_len) && (from_len <= bytes_in_rec));
+  return (compatible);
+}
+
+
+
 void Field_bit::sql_type(String &res) const
 {
   CHARSET_INFO *cs= res.charset();
diff --git a/sql/field.h b/sql/field.h
index bcdacd43082..55a95d3bb66 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -151,6 +151,24 @@ public:
     table, which is located on disk).
   */
   virtual uint32 pack_length_in_rec() const { return pack_length(); }
+  virtual int compatible_field_size(uint field_metadata);
+  virtual uint pack_length_from_metadata(uint field_metadata)
+  { return field_metadata; }
+  /*
+    This method is used to return the size of the data in a row-based
+    replication row record. The default implementation of returning 0 is
+    designed to allow fields that do not use metadata to return TRUE (1)
+    from compatible_field_size() which uses this function in the comparison.
+    The default value for field metadata for fields that do not have 
+    metadata is 0. Thus, 0 == 0 means the fields are compatible in size.
+
+    Note: While most classes that override this method return pack_length(),
+    the classes Field_string, Field_varstring, and Field_blob return 
+    field_length + 1, field_length, and pack_length_no_ptr() respectfully.
+  */
+  virtual uint row_pack_length() { return 0; }
+  virtual int save_field_metadata(uchar *first_byte)
+  { return do_save_field_metadata(first_byte); }
 
   /*
     data_length() return the "real size" of the data in memory.
@@ -463,6 +481,19 @@ private:
       overridden by subclasses.
    */
   virtual size_t do_last_null_byte() const;
+
+/**
+   Retrieve the field metadata for fields.
+
+   This default implementation returns 0 and saves 0 in the metadata_ptr
+   value.
+
+   @param   metadata_ptr   First byte of field metadata
+
+   @returns 0 no bytes written.
+*/
+  virtual int do_save_field_metadata(uchar *metadata_ptr)
+  { return 0; }
 };
 
 
@@ -589,6 +620,8 @@ public:
 
 /* New decimal/numeric field which use fixed point arithmetic */
 class Field_new_decimal :public Field_num {
+private:
+  int do_save_field_metadata(uchar *first_byte);
 public:
   /* The maximum number of decimal digits can be stored */
   uint precision;
@@ -628,6 +661,9 @@ public:
   uint32 max_display_length() { return field_length; }
   uint size_of() const { return sizeof(*this); } 
   uint32 pack_length() const { return (uint32) bin_size; }
+  uint pack_length_from_metadata(uint field_metadata);
+  uint row_pack_length() { return pack_length(); }
+  int compatible_field_size(uint field_metadata);
   uint is_equal(Create_field *new_field);
   virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
 };
@@ -834,7 +870,10 @@ public:
   int cmp(const uchar *,const uchar *);
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const { return sizeof(float); }
+  uint row_pack_length() { return pack_length(); }
   void sql_type(String &str) const;
+private:
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
@@ -871,7 +910,10 @@ public:
   int cmp(const uchar *,const uchar *);
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const { return sizeof(double); }
+  uint row_pack_length() { return pack_length(); }
   void sql_type(String &str) const;
+private:
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
@@ -1168,6 +1210,9 @@ public:
   uchar *pack(uchar *to, const uchar *from, uint max_length=~(uint) 0);
   virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
   const uchar *unpack(uchar* to, const uchar *from);
+  uint pack_length_from_metadata(uint field_metadata)
+  { return (field_metadata & 0x00ff); }
+  uint row_pack_length() { return (field_length + 1); }
   int pack_cmp(const uchar *a,const uchar *b,uint key_length,
                my_bool insert_or_update);
   int pack_cmp(const uchar *b,uint key_length,my_bool insert_or_update);
@@ -1179,6 +1224,8 @@ public:
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   Field *new_field(MEM_ROOT *root, struct st_table *new_table, bool keep_type);
   virtual uint get_key_image(uchar *buff,uint length, imagetype type);
+private:
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
@@ -1214,6 +1261,7 @@ public:
 
   enum_field_types type() const { return MYSQL_TYPE_VARCHAR; }
   enum ha_base_keytype key_type() const;
+  uint row_pack_length() { return field_length; }
   bool zero_pack() const { return 0; }
   int  reset(void) { bzero(ptr,field_length+length_bytes); return 0; }
   uint32 pack_length() const { return (uint32) field_length+length_bytes; }
@@ -1265,6 +1313,8 @@ public:
                        uint new_null_bit);
   uint is_equal(Create_field *new_field);
   void hash(ulong *nr, ulong *nr2);
+private:
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
@@ -1330,10 +1380,11 @@ public:
      This is used to determine the size of the actual data in the row
      buffer.
 
-     @retval The length of the raw data itself without the pointer.
+     @returns The length of the raw data itself without the pointer.
   */
   uint32 pack_length_no_ptr() const
   { return (uint32) (packlength); }
+  uint row_pack_length() { return pack_length_no_ptr(); }
   uint32 sort_length() const;
   inline uint32 max_data_length() const
   {
@@ -1360,7 +1411,7 @@ public:
      This is used to determine the size of the data plus the 
      packed length portion in the row data.
 
-     @retval The length in the row plus the size of the data.
+     @returns The length in the row plus the size of the data.
   */
   uint32 get_packed_size(const uchar *ptr_arg, bool low_byte_first)
     {return packlength + get_length(ptr_arg, low_byte_first);}
@@ -1430,6 +1481,8 @@ public:
   { return charset() == &my_charset_bin ? FALSE : TRUE; }
   uint32 max_display_length();
   uint is_equal(Create_field *new_field);
+private:
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
@@ -1499,12 +1552,17 @@ public:
   void sql_type(String &str) const;
   uint size_of() const { return sizeof(*this); }
   enum_field_types real_type() const { return MYSQL_TYPE_ENUM; }
+  uint pack_length_from_metadata(uint field_metadata)
+  { return (field_metadata & 0x00ff); }
+  uint row_pack_length() { return pack_length(); }
   virtual bool zero_pack() const { return 0; }
   bool optimize_range(uint idx, uint part) { return 0; }
   bool eq_def(Field *field);
   bool has_charset(void) const { return TRUE; }
   /* enum and set are sorted as integers */
   CHARSET_INFO *sort_charset(void) const { return &my_charset_bin; }
+private:
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
@@ -1525,6 +1583,7 @@ public:
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr) { return Field_set::store((longlong) nr, FALSE); }
   int  store(longlong nr, bool unsigned_val);
+
   virtual bool zero_pack() const { return 1; }
   String *val_str(String*,String *);
   void sql_type(String &str) const;
@@ -1588,6 +1647,10 @@ public:
   { get_key_image(buff, length, itRAW); }
   uint32 pack_length() const { return (uint32) (field_length + 7) / 8; }
   uint32 pack_length_in_rec() const { return bytes_in_rec; }
+  uint pack_length_from_metadata(uint field_metadata);
+  uint row_pack_length()
+  { return (bytes_in_rec + ((bit_len > 0) ? 1 : 0)); }
+  int compatible_field_size(uint field_metadata);
   void sql_type(String &str) const;
   uchar *pack(uchar *to, const uchar *from, uint max_length=~(uint) 0);
   virtual const uchar *unpack(uchar *to, const uchar *from, uint param_data);
@@ -1618,6 +1681,7 @@ public:
 
 private:
   virtual size_t do_last_null_byte() const;
+  int do_save_field_metadata(uchar *first_byte);
 };
 
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index bb8bc33fd82..fbb925fa98f 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -6384,63 +6384,18 @@ void Rows_log_event::print_helper(FILE *file,
 **************************************************************************/
 
 /**
-  * Calculate field metadata size based on the real_type of the field.
-  *
-  * @returns int Size of field metadata.
-  */
-#if !defined(MYSQL_CLIENT)
-const int Table_map_log_event::calc_field_metadata_size()
-{
-  DBUG_ENTER("Table_map_log_event::calc_field_metadata_size");
-  int size= 0;
-  for (unsigned int i= 0 ; i < m_table->s->fields ; i++)
-  {
-    switch (m_table->s->field[i]->real_type()) {
-    case MYSQL_TYPE_TINY_BLOB:
-    case MYSQL_TYPE_BLOB:
-    case MYSQL_TYPE_MEDIUM_BLOB:
-    case MYSQL_TYPE_LONG_BLOB:
-    case MYSQL_TYPE_DOUBLE:
-    case MYSQL_TYPE_FLOAT:
-    {
-      size++;                         // Store one byte here.
-      break; 
-    }
-    case MYSQL_TYPE_BIT:
-    case MYSQL_TYPE_NEWDECIMAL:
-    case MYSQL_TYPE_ENUM:
-    case MYSQL_TYPE_STRING:
-    case MYSQL_TYPE_VARCHAR:
-    case MYSQL_TYPE_SET:
-    {
-      size= size + 2; // Store short int here.
-      break;
-    }
-    default:
-      break;
-    }
-  }
-  m_field_metadata_size= size;
-  DBUG_PRINT("info", ("Table_map_log_event: %d bytes in field metadata.",
-                       (int)m_field_metadata_size));
-  DBUG_RETURN(m_field_metadata_size);
-}
-#endif /* !defined(MYSQL_CLIENT) */
-
-/**
   @page How replication of field metadata works.
   
   When a table map is created, the master first calls 
-  Table_map_log_event::get_field_metadata_size() which calculates how many 
+  Table_map_log_event::save_field_metadata() which calculates how many 
   values will be in the field metadata. Only those fields that require the 
-  extra data are added (see table above). The master then loops through all
-  of the fields in the table calling the method 
-  Table_map_log_event::get_field_metadata() which returns the values for the 
-  field that will be saved in the metadata and replicated to the slave. Once 
-  all fields have been processed, the table map is written to the binlog 
-  adding the size of the field metadata and the field metadata to the end of 
-  the body of the table map.
-  
+  extra data are added. The method also loops through all of the fields in 
+  the table calling the method Field::save_field_metadata() which returns the
+  values for the field that will be saved in the metadata and replicated to
+  the slave. Once all fields have been processed, the table map is written to
+  the binlog adding the size of the field metadata and the field metadata to
+  the end of the body of the table map.
+
   When a table map is read on the slave, the field metadata is read from the 
   table map and passed to the table_def class constructor which saves the 
   field metadata from the table map into an array based on the type of the 
@@ -6479,64 +6434,8 @@ int Table_map_log_event::save_field_metadata()
   DBUG_ENTER("Table_map_log_event::save_field_metadata");
   int index= 0;
   for (unsigned int i= 0 ; i < m_table->s->fields ; i++)
-  {
-    switch (m_table->s->field[i]->real_type()) {
-    case MYSQL_TYPE_NEWDECIMAL:
-    {
-      m_field_metadata[index++]= 
-        (uchar)((Field_new_decimal *)m_table->s->field[i])->precision;
-      m_field_metadata[index++]= 
-        (uchar)((Field_new_decimal *)m_table->s->field[i])->decimals();
-      break;
-    }
-    case MYSQL_TYPE_TINY_BLOB:
-    case MYSQL_TYPE_BLOB:
-    case MYSQL_TYPE_MEDIUM_BLOB:
-    case MYSQL_TYPE_LONG_BLOB:
-    {
-      m_field_metadata[index++]= 
-       (uchar)((Field_blob *)m_table->s->field[i])->pack_length_no_ptr();
-      break;
-    }
-    case MYSQL_TYPE_DOUBLE:
-    case MYSQL_TYPE_FLOAT:
-    {
-      m_field_metadata[index++]= (uchar)m_table->s->field[i]->pack_length();
-      break;
-    }
-    case MYSQL_TYPE_BIT:
-    { 
-      m_field_metadata[index++]= 
-        (uchar)((Field_bit *)m_table->s->field[i])->bit_len;
-      m_field_metadata[index++]= 
-        (uchar)((Field_bit *)m_table->s->field[i])->bytes_in_rec;
-      break;
-    }
-    case MYSQL_TYPE_VARCHAR:
-    {
-      char *ptr= (char *)&m_field_metadata[index];
-      int2store(ptr, m_table->s->field[i]->field_length);
-      index= index + 2;
-      break;
-    }
-    case MYSQL_TYPE_STRING:
-    {
-      m_field_metadata[index++]= (uchar)m_table->s->field[i]->real_type();
-      m_field_metadata[index++]= m_table->s->field[i]->field_length;
-      break;
-    }
-    case MYSQL_TYPE_ENUM:
-    case MYSQL_TYPE_SET:
-    {
-      m_field_metadata[index++]= (uchar)m_table->s->field[i]->real_type();
-      m_field_metadata[index++]= m_table->s->field[i]->pack_length();
-      break;
-    }
-    default:
-      break;
-    }
-  }
-  DBUG_RETURN(0);
+    index+= m_table->s->field[i]->save_field_metadata(&m_field_metadata[index]);
+  DBUG_RETURN(index);
 }
 #endif /* !defined(MYSQL_CLIENT) */
 
@@ -6574,16 +6473,6 @@ Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
   m_data_size+= m_dblen + 2;	// Include length and terminating \0
   m_data_size+= m_tbllen + 2;	// Include length and terminating \0
   m_data_size+= 1 + m_colcnt;	// COLCNT and column types
-  m_field_metadata_size= calc_field_metadata_size();
-
-  /*
-    Now set the size of the data to the size of the field metadata array
-    plus one or two bytes for number of elements in the field metadata array.
-  */
-  if (m_field_metadata_size > 255)
-    m_data_size+= m_field_metadata_size + 2; 
-  else
-    m_data_size+= m_field_metadata_size + 1; 
 
   /* If malloc fails, catched in is_valid() */
   if ((m_memory= (uchar*) my_malloc(m_colcnt, MYF(MY_WME))))
@@ -6603,17 +6492,31 @@ Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
   m_data_size+= num_null_bytes;
   m_meta_memory= (uchar *)my_multi_malloc(MYF(MY_WME),
                                  &m_null_bits, num_null_bytes,
-                                 &m_field_metadata, m_field_metadata_size,
+                                 &m_field_metadata, (m_colcnt * 2),
                                  NULL);
+
+  bzero(m_field_metadata, (m_colcnt * 2));
+
+  /*
+    Create an array for the field metadata and store it.
+  */
+  m_field_metadata_size= save_field_metadata();
+  DBUG_ASSERT(m_field_metadata_size <= (m_colcnt * 2));
+
+  /*
+    Now set the size of the data to the size of the field metadata array
+    plus one or two bytes for number of elements in the field metadata array.
+  */
+  if (m_field_metadata_size > 255)
+    m_data_size+= m_field_metadata_size + 2; 
+  else
+    m_data_size+= m_field_metadata_size + 1; 
+
   bzero(m_null_bits, num_null_bytes);
   for (unsigned int i= 0 ; i < m_table->s->fields ; ++i)
     if (m_table->field[i]->maybe_null())
       m_null_bits[(i / 8)]+= 1 << (i % 8);
 
-  /*
-    Create an array for the field metadata and store it.
-  */
-  save_field_metadata();
 }
 #endif /* !defined(MYSQL_CLIENT) */
 
diff --git a/sql/log_event.h b/sql/log_event.h
index e22a9785736..fa22c875c48 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -2049,7 +2049,6 @@ public:
 
   virtual int get_data_size() { return m_data_size; } 
 #ifndef MYSQL_CLIENT
-  virtual const int calc_field_metadata_size();
   virtual int save_field_metadata();
   virtual bool write_data_header(IO_CACHE *file);
   virtual bool write_data_body(IO_CACHE *file);
@@ -2090,7 +2089,7 @@ private:
 
   uchar          *m_field_metadata;        // buffer for field metadata
   /*
-    The size of field metadata buffer set by calling calc_field_metadata_size()
+    The size of field metadata buffer set by calling save_field_metadata()
   */
   ulong          m_field_metadata_size;   
   uchar         *m_null_bits;
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index 85ed6089f56..25daf1471ce 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -191,6 +191,25 @@ table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table)
       rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
                   ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
     }
+    /*
+      Check the slave's field size against that of the master.
+    */
+    if (!error && 
+        !table->field[col]->compatible_field_size(field_metadata(col)))
+    {
+      error= 1;
+      char buf[256];
+      my_snprintf(buf, sizeof(buf), "Column %d size mismatch - "
+                  "master has size %d, %s.%s on slave has size %d."
+                  " Master's column size should be <= the slave's "
+                  "column size.", col,
+                  table->field[col]->pack_length_from_metadata(
+                                       m_field_metadata[col]),
+                  tsh->db.str, tsh->table_name.str, 
+                  table->field[col]->row_pack_length());
+      rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
+                  ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
+    }
   }
 
   return error;
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index d08f62363af..e0e78bf402b 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -61,7 +61,7 @@ public:
    */
   table_def(field_type *types, ulong size, uchar *field_metadata, 
       int metadata_size, uchar *null_bitmap)
-    : m_size(size), m_type(0),
+    : m_size(size), m_type(0), m_field_metadata_size(metadata_size),
       m_field_metadata(0), m_null_bits(0), m_memory(NULL)
   {
     m_memory= (uchar *)my_multi_malloc(MYF(MY_WME),
@@ -194,7 +194,7 @@ public:
   uint16 field_metadata(uint index) const
   {
     DBUG_ASSERT(index < m_size);
-    if (m_field_metadata)
+    if (m_field_metadata_size)
       return m_field_metadata[index];
     else
       return 0;
@@ -243,6 +243,7 @@ private:
   ulong m_size;           // Number of elements in the types array
   field_type *m_type;                     // Array of type descriptors
   uint16 *m_field_metadata;
+  uint m_field_metadata_size;
   uchar *m_null_bits;
   uchar *m_memory;
 };