WL#3228 (NDB) : RBR using different table defs on slave/master

This patch adds the ability to store extra field metadata in the table
map event. This data can include pack_length() or field_lenght() for
fields such as CHAR or VARCHAR enabling developers to add code that
can check for compatibilty between master and slave columns. More 
importantly, the extra field metadata can be used to store data from the
master correctly should a VARCHAR field on the master be <= 255 bytes 
while the same field on the slave is > 255 bytes. 

The patch also includes the needed changes to unpack to ensure that data
which is smaller on the master can be unpacked correctly on the slave.

WL#3915 : (NDB) master's cols > slave

Slave starts accepting and handling rows of master's tables which have more columns.
The most important part of implementation is how to caclulate the amount of bytes to
skip for unknown by slave column.

8 files changed, 737 insertions, 87 deletions

diff --git a/sql/field.cc b/sql/field.cc
index a970a6e4318..49433deca74 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -1372,6 +1372,51 @@ int Field::store(const char *to, uint length, CHARSET_INFO *cs,
 }
 
 
+/**
+   Unpack a field from row data.
+
+   This method is used to unpack a field from a master whose size 
+   of the field is less than that of the slave.
+  
+   @param   to         Destination of the data
+   @param   from       Source of the data
+   @param   param_data Pack length of the field data
+
+   @return  New pointer into memory based on from + length of the data
+*/
+const uchar *Field::unpack(uchar* to,
+                           const uchar *from, 
+                           uint param_data)
+{
+  uint length=pack_length();
+  int from_type= 0;
+  /*
+    If from length is > 255, it has encoded data in the upper bits. Need
+    to mask it out.
+  */
+  if (param_data > 255)
+  {
+    from_type= (param_data & 0xff00) >> 8U;  // real_type.
+    param_data= param_data & 0x00ff;        // length.
+  }
+  uint len= (param_data && (param_data < length)) ?
+            param_data : length;
+  /*
+    If the length is the same, use old unpack method.
+    If the param_data is 0, use the old unpack method.
+      This is possible if the table map was generated from a down-level
+      master or if the data was not available on the master.
+    If the real_types are not the same, use the old unpack method.
+  */
+  if ((length == param_data) ||
+      (param_data == 0) ||
+      (from_type != real_type()))
+    return(unpack(to, from));
+  memcpy(to, from, param_data > length ? length : len);
+  return from+len;
+}
+
+
 my_decimal *Field::val_decimal(my_decimal *decimal)
 {
   /* This never have to be called */
@@ -2657,6 +2702,52 @@ uint Field_new_decimal::is_equal(Create_field *new_field)
 }
 
 
+/**
+   Unpack a decimal field from row data.
+
+   This method is used to unpack a decimal or numeric field from a master
+   whose size of the field is less than that of the slave.
+  
+   @param   to         Destination of the data
+   @param   from       Source of the data
+   @param   param_data Precision (upper) and decimal (lower) values
+
+   @return  New pointer into memory based on from + length of the data
+*/
+const uchar *Field_new_decimal::unpack(uchar* to, 
+                                       const uchar *from, 
+                                       uint param_data)
+{
+  uint from_precision= (param_data & 0xff00) >> 8U;
+  uint from_decimal= param_data & 0x00ff;
+  uint length=pack_length();
+  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 the master's data is smaller than the slave, we need to convert
+      the binary to decimal then resize the decimal converting it back to
+      a decimal and write that to the raw data buffer.
+    */
+    decimal_digit_t dec_buf[DECIMAL_MAX_PRECISION];
+    decimal_t dec;
+    dec.len= from_precision;
+    dec.buf= dec_buf;
+    /*
+      Note: bin2decimal does not change the length of the field. So it is
+      just the first step the resizing operation. The second step does the
+      resizing using the precision and decimals from the slave.
+    */
+    bin2decimal((uchar *)from, &dec, from_precision, from_decimal);
+    decimal2bin(&dec, to, precision, decimals());
+  }
+  else
+    memcpy(to, from, len); // Sizes are the same, just copy the data.
+  return from+len;
+}
+
 /****************************************************************************
 ** tiny int
 ****************************************************************************/
@@ -6306,6 +6397,38 @@ uchar *Field_string::pack(uchar *to, const uchar *from, uint max_length)
 }
 
 
+/**
+   Unpack a string field from row data.
+
+   This method is used to unpack a string field from a master whose size 
+   of the field is less than that of the slave. Note that there can be a
+   variety of field types represented with this class. Certain types like
+   ENUM or SET are processed differently. Hence, the upper byte of the 
+   @c param_data argument contains the result of field->real_type() from
+   the master.
+
+   @param   to         Destination of the data
+   @param   from       Source of the data
+   @param   param_data Real type (upper) and length (lower) values
+
+   @return  New pointer into memory based on from + length of the data
+*/
+const uchar *Field_string::unpack(uchar *to,
+                                  const uchar *from,
+                                  uint param_data)
+{
+  uint from_len= param_data & 0x00ff;                 // length.
+  uint length= 0;
+  uint f_length;
+  f_length= (from_len < field_length) ? from_len : field_length;
+  DBUG_ASSERT(f_length <= 255);
+  length= (uint) *from++;
+  bitmap_set_bit(table->write_set,field_index);
+  store((const char *)from, length, system_charset_info);
+  return from+length;
+}
+
+
 const uchar *Field_string::unpack(uchar *to, const uchar *from)
 {
   uint length;
@@ -6797,6 +6920,44 @@ uchar *Field_varstring::pack_key_from_key_image(uchar *to, const uchar *from,
 }
 
 
+/**
+   Unpack a varstring field from row data.
+
+   This method is used to unpack a varstring field from a master
+   whose size of the field is less than that of the slave.
+  
+   @param   to         Destination of the data
+   @param   from       Source of the data
+   @param   param_data Length bytes from the master's field data
+
+   @return  New pointer into memory based on from + length of the data
+*/
+const uchar *Field_varstring::unpack(uchar *to, 
+                                     const uchar *from,
+                                     uint param_data)
+{
+  uint length;
+  uint l_bytes= (param_data && (param_data < field_length)) ? 
+                (param_data <= 255) ? 1 : 2 : length_bytes;
+  if (l_bytes == 1)
+  {
+    to[0]= *from++;
+    length= to[0];
+    if (length_bytes == 2)
+      to[1]= 0;
+  }
+  else
+  {
+    length= uint2korr(from);
+    to[0]= *from++;
+    to[1]= *from++;
+  }
+  if (length)
+    memcpy(to+ length_bytes, from, length);
+  return from+length;
+}
+
+
 /*
   unpack field packed with Field_varstring::pack()
 */
@@ -7493,6 +7654,29 @@ uchar *Field_blob::pack(uchar *to, const uchar *from, uint max_length)
 }
 
 
+/**
+   Unpack a blob field from row data.
+
+   This method is used to unpack a blob field from a master whose size of 
+   the field is less than that of the slave. Note: This method is included
+   to satisfy inheritance rules, but is not needed for blob fields. It
+   simply is used as a pass-through to the original unpack() method for
+   blob fields.
+
+   @param   to         Destination of the data
+   @param   from       Source of the data
+   @param   param_data <not used>
+
+   @return  New pointer into memory based on from + length of the data
+*/
+const uchar *Field_blob::unpack(uchar *to, 
+                                const uchar *from,
+                                uint param_data)
+{
+  return unpack(to, from);
+}
+
+
 const uchar *Field_blob::unpack(uchar *to, const uchar *from)
 {
   memcpy(to,from,packlength);
@@ -8540,6 +8724,58 @@ uchar *Field_bit::pack(uchar *to, const uchar *from, uint max_length)
 }
 
 
+/**
+   Unpack a bit field from row data.
+
+   This method is used to unpack a bit field from a master whose size 
+   of the field is less than that of the slave.
+  
+   @param   to         Destination of the data
+   @param   from       Source of the data
+   @param   param_data Bit length (upper) and length (lower) values
+
+   @return  New pointer into memory based on from + length of the data
+*/
+const uchar *Field_bit::unpack(uchar *to,
+                               const uchar *from,
+                               uint param_data)
+{
+  uint const from_len= (param_data >> 8U) & 0x00ff;
+  uint const from_bit_len= param_data & 0x00ff;
+  /*
+    If the master and slave have the same sizes, then use the old
+    unpack() method.
+  */
+  if ((from_bit_len == bit_len) &&
+      (from_len == bytes_in_rec)) 
+    return(unpack(to, from));
+  /*
+    We are converting a smaller bit field to a larger one here.
+    To do that, we first need to construct a raw value for the original
+    bit value stored in the from buffer. Then that needs to be converted
+    to the larger field then sent to store() for writing to the field.
+    Lastly the odd bits need to be masked out if the bytes_in_rec > 0.
+    Otherwise stray bits can cause spurious values.
+  */
+  uint new_len= (field_length + 7) / 8;
+  char *value= (char *)my_alloca(new_len);
+  bzero(value, new_len);
+  uint len= from_len + ((from_bit_len > 0) ? 1 : 0);
+  memcpy(value + (new_len - len), from, len);
+  /*
+    Mask out the unused bits in the partial byte. 
+    TODO: Add code to the master to always mask these bits and remove
+          the following.
+  */
+  if ((from_bit_len > 0) && (from_len > 0))
+    value[new_len - len]= value[new_len - len] & ((1U << from_bit_len) - 1);
+  bitmap_set_bit(table->write_set,field_index);
+  store(value, new_len, system_charset_info);
+  my_afree(value);
+  return from + len;
+}
+
+
 const uchar *Field_bit::unpack(uchar *to, const uchar *from)
 {
   if (bit_len > 0)
diff --git a/sql/field.h b/sql/field.h
index 8bf087c7ebd..c1130d53569 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -345,6 +345,7 @@ public:
     memcpy(to,from,length);
     return to+length;
   }
+  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
   virtual const uchar *unpack(uchar* to, const uchar *from)
   {
     uint length=pack_length();
@@ -628,6 +629,7 @@ public:
   uint size_of() const { return sizeof(*this); } 
   uint32 pack_length() const { return (uint32) bin_size; }
   uint is_equal(Create_field *new_field);
+  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
 };
 
 
@@ -1164,6 +1166,7 @@ public:
   void sort_string(uchar *buff,uint length);
   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);
   const uchar *unpack(uchar* to, const uchar *from);
   int pack_cmp(const uchar *a,const uchar *b,uint key_length,
                my_bool insert_or_update);
@@ -1240,6 +1243,7 @@ public:
   uchar *pack_key(uchar *to, const uchar *from, uint max_length);
   uchar *pack_key_from_key_image(uchar* to, const uchar *from,
                                  uint max_length);
+  virtual const uchar *unpack(uchar* to, const uchar *from, uint param_data);
   const uchar *unpack(uchar* to, const uchar *from);
   const uchar *unpack_key(uchar* to, const uchar *from, uint max_length);
   int pack_cmp(const uchar *a, const uchar *b, uint key_length,
@@ -1295,6 +1299,9 @@ public:
                   l_char_length <= 16777215 ? 3 : 4;
     }
   }
+  Field_blob(uint32 packlength_arg)
+    :Field_longstr((uchar*) 0, 0, (uchar*) "", 0, NONE, "temp", system_charset_info),
+    packlength(packlength_arg) {}
   enum_field_types type() const { return MYSQL_TYPE_BLOB;}
   enum ha_base_keytype key_type() const
     { return binary() ? HA_KEYTYPE_VARBINARY2 : HA_KEYTYPE_VARTEXT2; }
@@ -1316,6 +1323,17 @@ public:
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const
   { return (uint32) (packlength+table->s->blob_ptr_size); }
+
+  /**
+     Return the packed length without the pointer size added. 
+
+     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.
+  */
+  uint32 pack_length_no_ptr() const
+  { return (uint32) (packlength); }
   uint32 sort_length() const;
   inline uint32 max_data_length() const
   {
@@ -1331,7 +1349,18 @@ public:
   {
     store_length(ptr, packlength, number);
   }
- 
+
+  /**
+     Return the packed length plus the length of the data. 
+
+     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.
+  */
+  uint32 get_packed_size(const uchar *ptr)
+    {return packlength + get_length((const uchar *)ptr);}
+
   inline uint32 get_length(uint row_offset=0)
   { return get_length(ptr+row_offset); }
   uint32 get_length(const uchar *ptr);
@@ -1379,6 +1408,7 @@ public:
   uchar *pack_key(uchar *to, const uchar *from, uint max_length);
   uchar *pack_key_from_key_image(uchar* to, const uchar *from,
                                  uint max_length);
+  virtual const uchar *unpack(uchar *to, const uchar *from, uint param_data);
   const uchar *unpack(uchar *to, const uchar *from);
   const uchar *unpack_key(uchar* to, const uchar *from, uint max_length);
   int pack_cmp(const uchar *a, const uchar *b, uint key_length,
@@ -1554,6 +1584,7 @@ public:
   uint32 pack_length_in_rec() const { return bytes_in_rec; }
   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);
   const uchar *unpack(uchar* to, const uchar *from);
   virtual void set_default();
 
diff --git a/sql/log_event.cc b/sql/log_event.cc
index dc492d08c10..22b5a96753d 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -6028,11 +6028,8 @@ int Rows_log_event::do_apply_event(RELAY_LOG_INFO const *rli)
 #ifdef HAVE_QUERY_CACHE
     query_cache.invalidate_locked_for_write(rli->tables_to_lock);
 #endif
-    const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock();
   }
 
-  DBUG_ASSERT(rli->tables_to_lock == NULL && rli->tables_to_lock_count == 0);
-
   TABLE* table= const_cast<RELAY_LOG_INFO*>(rli)->m_table_map.get_table(m_table_id);
 
   if (table)
@@ -6128,6 +6125,13 @@ int Rows_log_event::do_apply_event(RELAY_LOG_INFO const *rli)
     }
   }
 
+  /*
+    We need to delay this clear until the table def is no longer needed.
+    The table def is needed in unpack_row().
+  */
+  if (rli->tables_to_lock && get_flags(STMT_END_F))
+    const_cast<RELAY_LOG_INFO*>(rli)->clear_tables_to_lock();
+
   if (error)
   {                     /* error has occured during the transaction */
     rli->report(ERROR_LEVEL, thd->net.last_errno,
@@ -6378,6 +6382,163 @@ void Rows_log_event::print_helper(FILE *file,
 	Table_map_log_event member functions and support functions
 **************************************************************************/
 
+/**
+  * 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 + sizeof(short int); // 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 
+  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.
+  
+  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 
+  field. Field metadata values not present (those fields that do not use extra 
+  data) in the table map are initialized as zero (0). The array size is the 
+  same as the columns for the table on the slave.
+
+*/
+
+/**
+  Save the field metadata based on the real_type of the field.
+  The metadata saved depends on the type of the field. Some fields
+  store a single byte for pack_length() while others store two bytes
+  for field_length (max length).
+  
+  @retval 0 Ok.
+
+  TODO: We may want to consider changing the encoding of the information.
+  Currently, the code attempts to minimize the number of bytes written to 
+  the tablemap. There are at least two other alternatives; 1) using 
+  net_store_length() to store the data allowing it to choose the number of
+  bytes that are appropriate thereby making the code much easier to 
+  maintain (only 1 place to change the encoding), or 2) use a fixed number
+  of bytes for each field. The problem with option 1 is that net_store_length()
+  will use one byte if the value < 251, but 3 bytes if it is > 250. Thus,
+  for fields like CHAR which can be no larger than 255 characters, the method
+  will use 3 bytes when the value is > 250. Further, every value that is
+  encoded using 2 parts (e.g., pack_length, field_length) will be numerically
+  > 250 therefore will use 3 bytes for eah value. The problem with option 2
+  is less wasteful for space but does waste 1 byte for every field that does
+  not encode 2 parts. 
+*/
+#if !defined(MYSQL_CLIENT)
+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 + sizeof(short int);
+      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);
+}
+#endif /* !defined(MYSQL_CLIENT) */
+
 /*
   Constructor used to build an event for writing to the binary log.
   Mats says tbl->s lives longer than this event so it's ok to copy pointers
@@ -6392,9 +6553,8 @@ Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
     m_dblen(m_dbnam ? tbl->s->db.length : 0),
     m_tblnam(tbl->s->table_name.str),
     m_tbllen(tbl->s->table_name.length),
-    m_colcnt(tbl->s->fields), m_coltype(0),
-    m_table_id(tid),
-    m_flags(flags)
+    m_colcnt(tbl->s->fields), m_field_metadata(0),
+    m_table_id(tid), m_null_bits(0), m_flags(flags)
 {
   DBUG_ASSERT(m_table_id != ~0UL);
   /*
@@ -6413,6 +6573,16 @@ 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))))
@@ -6421,6 +6591,28 @@ Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
     for (unsigned int i= 0 ; i < m_table->s->fields ; ++i)
       m_coltype[i]= m_table->field[i]->type();
   }
+
+  /*
+    Calculate a bitmap for the results of maybe_null() for all columns.
+    The bitmap is used to determine when there is a column from the master
+    that is not on the slave and is null and thus not in the row data during
+    replication.
+  */
+  uint num_null_bytes= (m_table->s->fields + 7) / 8;
+  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,
+                                 NULL);
+  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) */
 
@@ -6436,8 +6628,10 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len,
 #ifndef MYSQL_CLIENT
   m_table(NULL),
 #endif
-  m_memory(NULL)
+  m_memory(NULL),
+  m_field_metadata(0), m_field_metadata_size(0)
 {
+  unsigned int bytes_read= 0;
   DBUG_ENTER("Table_map_log_event::Table_map_log_event(const char*,uint,...)");
 
   uint8 common_header_len= description_event->common_header_len;
@@ -6508,6 +6702,22 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len,
     strncpy(const_cast<char*>(m_dbnam), (const char*)ptr_dblen  + 1, m_dblen + 1);
     strncpy(const_cast<char*>(m_tblnam), (const char*)ptr_tbllen + 1, m_tbllen + 1);
     memcpy(m_coltype, ptr_after_colcnt, m_colcnt);
+
+    ptr_after_colcnt= ptr_after_colcnt + m_colcnt;
+    bytes_read= ptr_after_colcnt - (uchar *)buf;
+    DBUG_PRINT("info", ("Bytes read: %d.\n", bytes_read));
+    if (bytes_read < event_len)
+    {
+      m_field_metadata_size= net_field_length(&ptr_after_colcnt);
+      uint num_null_bytes= (m_colcnt + 7) / 8;
+      m_meta_memory= (uchar *)my_multi_malloc(MYF(MY_WME),
+                                     &m_null_bits, num_null_bytes,
+                                     &m_field_metadata, m_field_metadata_size,
+                                     NULL);
+      memcpy(m_field_metadata, ptr_after_colcnt, m_field_metadata_size);
+      ptr_after_colcnt= (uchar*)ptr_after_colcnt + m_field_metadata_size;
+      memcpy(m_null_bits, ptr_after_colcnt, num_null_bytes);
+    }
   }
 
   DBUG_VOID_RETURN;
@@ -6516,6 +6726,7 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len,
 
 Table_map_log_event::~Table_map_log_event()
 {
+  my_free(m_meta_memory, MYF(MY_ALLOW_ZERO_PTR));
   my_free(m_memory, MYF(MY_ALLOW_ZERO_PTR));
 }
 
@@ -6646,7 +6857,8 @@ int Table_map_log_event::do_apply_event(RELAY_LOG_INFO const *rli)
       inside st_relay_log_info::clear_tables_to_lock() by calling the
       table_def destructor explicitly.
     */
-    new (&table_list->m_tabledef) table_def(m_coltype, m_colcnt);
+    new (&table_list->m_tabledef) table_def(m_coltype, m_colcnt, 
+         m_field_metadata, m_field_metadata_size, m_null_bits);
     table_list->m_tabledef_valid= TRUE;
 
     /*
@@ -6718,12 +6930,22 @@ bool Table_map_log_event::write_data_body(IO_CACHE *file)
   uchar *const cbuf_end= net_store_length(cbuf, (size_t) m_colcnt);
   DBUG_ASSERT(static_cast<size_t>(cbuf_end - cbuf) <= sizeof(cbuf));
 
+  /*
+    Store the size of the field metadata.
+  */
+  uchar mbuf[sizeof(m_field_metadata_size)];
+  uchar *const mbuf_end= net_store_length(mbuf, 
+                                         (size_t) m_field_metadata_size);
+
   return (my_b_safe_write(file, dbuf,      sizeof(dbuf)) ||
           my_b_safe_write(file, (const uchar*)m_dbnam,   m_dblen+1) ||
           my_b_safe_write(file, tbuf,      sizeof(tbuf)) ||
           my_b_safe_write(file, (const uchar*)m_tblnam,  m_tbllen+1) ||
           my_b_safe_write(file, cbuf, (size_t) (cbuf_end - cbuf)) ||
-          my_b_safe_write(file, m_coltype, m_colcnt));
+          my_b_safe_write(file, m_coltype, m_colcnt) ||
+          my_b_safe_write(file, mbuf, (size_t) (mbuf_end - mbuf)) ||
+          my_b_safe_write(file, m_field_metadata, m_field_metadata_size),
+          my_b_safe_write(file, m_null_bits, (m_colcnt + 7) / 8));
  }
 #endif
 
@@ -6946,6 +7168,7 @@ copy_extra_record_fields(TABLE *table,
                          size_t master_reclength,
                          my_ptrdiff_t master_fields)
 {
+  DBUG_ENTER("copy_extra_record_fields(table, master_reclen, master_fields)");
   DBUG_PRINT("info", ("Copying to 0x%lx "
                       "from field %lu at offset %lu "
                       "to field %d at offset %lu",
@@ -6956,7 +7179,11 @@ copy_extra_record_fields(TABLE *table,
     Copying the extra fields of the slave that does not exist on
     master into record[0] (which are basically the default values).
   */
-  DBUG_ASSERT(master_reclength <= table->s->reclength);
+
+  if (table->s->fields < (uint) master_fields)
+    DBUG_RETURN(0);
+
+ DBUG_ASSERT(master_reclength <= table->s->reclength);
   if (master_reclength < table->s->reclength)
     bmove_align(table->record[0] + master_reclength,
                 table->record[1] + master_reclength,
@@ -7013,7 +7240,7 @@ copy_extra_record_fields(TABLE *table,
       }
     }
   }
-  return 0;                                     // All OK
+  DBUG_RETURN(0);                                     // All OK
 }
 
 #define DBUG_PRINT_BITSET(N,FRM,BS)                \
@@ -7603,12 +7830,6 @@ int Delete_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO const *rli,
                                           uchar const **const row_end)
 {
   DBUG_ASSERT(row_start && row_end);
-  /*
-    This assertion actually checks that there is at least as many
-    columns on the slave as on the master.
-  */
-  DBUG_ASSERT(table->s->fields >= m_width);
-
   if (int error= unpack_row(rli, table, m_width, row_start, &m_cols, row_end,
                             &m_master_reclength, table->read_set, DELETE_ROWS_EVENT))
   {
@@ -7787,12 +8008,6 @@ int Update_rows_log_event::do_prepare_row(THD *thd, RELAY_LOG_INFO const *rli,
   int error;
   DBUG_ASSERT(row_start && row_end);
   /*
-    This assertion actually checks that there is at least as many
-    columns on the slave as on the master.
-  */
-  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().
diff --git a/sql/log_event.h b/sql/log_event.h
index 6787aa08016..e22a9785736 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -2049,6 +2049,8 @@ 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);
   virtual const char *get_db() { return m_dbnam; }
@@ -2085,6 +2087,14 @@ private:
   flag_set       m_flags;
 
   size_t         m_data_size;
+
+  uchar          *m_field_metadata;        // buffer for field metadata
+  /*
+    The size of field metadata buffer set by calling calc_field_metadata_size()
+  */
+  ulong          m_field_metadata_size;   
+  uchar         *m_null_bits;
+  uchar         *m_meta_memory;
 };
 
 
diff --git a/sql/rpl_record.cc b/sql/rpl_record.cc
index 94778948bf9..094ccbcdd7d 100644
--- a/sql/rpl_record.cc
+++ b/sql/rpl_record.cc
@@ -17,6 +17,8 @@
 #include "rpl_rli.h"
 #include "rpl_record.h"
 #include "slave.h"                  // Need to pull in slave_print_msg
+#include "rpl_utility.h"
+#include "rpl_rli.h"
 
 /**
    Pack a record of data for a table into a format suitable for
@@ -143,7 +145,8 @@ pack_row(TABLE *table, MY_BITMAP const* cols,
    @param rli     Relay log info
    @param table   Table to unpack into
    @param colcnt  Number of columns to read from record
-   @param row     Packed row data
+   @param row_data
+                  Packed row data
    @param cols    Pointer to columns data to fill in
    @param row_end Pointer to variable that will hold the value of the
                   one-after-end position for the row
@@ -191,7 +194,9 @@ unpack_row(RELAY_LOG_INFO const *rli,
   unsigned int null_mask= 1U;
   // The "current" null bits
   unsigned int null_bits= *null_ptr++;
-  for (field_ptr= begin_ptr ; field_ptr < end_ptr ; ++field_ptr)
+  uint i= 0;
+  table_def *tabledef= ((RELAY_LOG_INFO*)rli)->get_tabledef(table);
+  for (field_ptr= begin_ptr ; field_ptr < end_ptr && *field_ptr ; ++field_ptr)
   {
     Field *const f= *field_ptr;
 
@@ -220,14 +225,42 @@ unpack_row(RELAY_LOG_INFO const *rli,
         f->set_notnull();
 
         /*
-          We only unpack the field if it was non-null
-        */
-        pack_ptr= f->unpack(f->ptr, pack_ptr);
+          We only unpack the field if it was non-null.
+          Use the master's size information if available else call
+          normal unpack operation.
+         */
+        if (tabledef && tabledef->field_metadata(i))
+          pack_ptr= f->unpack(f->ptr, pack_ptr, tabledef->field_metadata(i));
+        else
+          pack_ptr= f->unpack(f->ptr, pack_ptr);
       }
 
       bitmap_set_bit(rw_set, f->field_index);
       null_mask <<= 1;
     }
+    i++;
+  }
+
+  /*
+    throw away master's extra fields
+  */
+  uint max_cols= min(tabledef->size(), cols->n_bits);
+  for (; i < max_cols; i++)
+  {
+    if (bitmap_is_set(cols, i))
+    {
+      if ((null_mask & 0xFF) == 0)
+      {
+        DBUG_ASSERT(null_ptr < row_data + master_null_byte_count);
+        null_mask= 1U;
+        null_bits= *null_ptr++;
+      }
+      DBUG_ASSERT(null_mask & 0xFF); // One of the 8 LSB should be set
+
+      if (!((null_bits & null_mask) && tabledef->maybe_null(i)))
+        pack_ptr+= tabledef->calc_field_size(i, (uchar *) pack_ptr);
+      null_mask <<= 1;
+    }
   }
 
   /*
diff --git a/sql/rpl_rli.h b/sql/rpl_rli.h
index c458318594a..7ae21897e3f 100644
--- a/sql/rpl_rli.h
+++ b/sql/rpl_rli.h
@@ -18,6 +18,7 @@
 
 #include "rpl_tblmap.h"
 #include "rpl_reporting.h"
+#include "rpl_utility.h"
 
 struct RPL_TABLE_LIST;
 
@@ -301,6 +302,15 @@ typedef struct st_relay_log_info : public Slave_reporting_capability
   uint tables_to_lock_count;        /* RBR: Count of tables to lock */
   table_mapping m_table_map;      /* RBR: Mapping table-id to table */
 
+  inline table_def *get_tabledef(TABLE *tbl)
+  {
+    table_def *td= 0;
+    for (TABLE_LIST *ptr= tables_to_lock; ptr && !td; ptr= ptr->next_global)
+      if (ptr->table == tbl)
+        td= &((RPL_TABLE_LIST *)ptr)->m_tabledef;
+    return (td);
+  }
+
   /*
     Last charset (6 bytes) seen by slave SQL thread is cached here; it helps
     the thread save 3 get_charset() per Query_log_event if the charset is not
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index a04bcd1fab9..2a124c7b14e 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -16,17 +16,45 @@
 #include "rpl_utility.h"
 #include "rpl_rli.h"
 
-uint32
-field_length_from_packed(enum_field_types const field_type, 
-                         uchar const *const data)
+/*********************************************************************
+ *                   table_def member definitions                    *
+ *********************************************************************/
+
+/*
+  This function returns the field size in raw bytes based on the type
+  and the encoded field data from the master's raw data.
+*/
+uint32 table_def::calc_field_size(uint col, uchar *master_data)
 {
   uint32 length;
 
-  switch (field_type) {
-  case MYSQL_TYPE_DECIMAL:
+  switch (type(col)) {
   case MYSQL_TYPE_NEWDECIMAL:
-    length= ~(uint32) 0;
+    length= my_decimal_get_binary_size(m_field_metadata[col] >> 8, 
+             m_field_metadata[col] - ((m_field_metadata[col] >> 8) << 8));
+    break;
+  case MYSQL_TYPE_DECIMAL:
+  case MYSQL_TYPE_FLOAT:
+  case MYSQL_TYPE_DOUBLE:
+    length= m_field_metadata[col];
+    break;
+  case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_ENUM:
+  case MYSQL_TYPE_STRING:
+  {
+    if (((m_field_metadata[col] & 0xff00) == (MYSQL_TYPE_SET << 8)) ||
+        ((m_field_metadata[col] & 0xff00) == (MYSQL_TYPE_ENUM << 8)))
+      length= m_field_metadata[col] & 0x00ff;
+    else
+    {
+      length= m_field_metadata[col] & 0x00ff;
+      if (length > 255)
+        length= uint2korr(master_data) + 2;
+      else
+        length= (uint) *master_data + 1;
+    }
     break;
+  }
   case MYSQL_TYPE_YEAR:
   case MYSQL_TYPE_TINY:
     length= 1;
@@ -45,12 +73,6 @@ field_length_from_packed(enum_field_types const field_type,
     length= 8;
     break;
 #endif
-  case MYSQL_TYPE_FLOAT:
-    length= sizeof(float);
-    break;
-  case MYSQL_TYPE_DOUBLE:
-    length= sizeof(double);
-    break;
   case MYSQL_TYPE_NULL:
     length= 0;
     break;
@@ -58,8 +80,6 @@ field_length_from_packed(enum_field_types const field_type,
     length= 3;
     break;
   case MYSQL_TYPE_DATE:
-    length= 4;
-    break;
   case MYSQL_TYPE_TIME:
     length= 3;
     break;
@@ -69,41 +89,33 @@ field_length_from_packed(enum_field_types const field_type,
   case MYSQL_TYPE_DATETIME:
     length= 8;
     break;
-    break;
   case MYSQL_TYPE_BIT:
-    length= ~(uint32) 0;
-    break;
-  default:
-    /* This case should never be chosen */
-    DBUG_ASSERT(0);
-    /* If something goes awfully wrong, it's better to get a string than die */
-  case MYSQL_TYPE_STRING:
-    length= uint2korr(data);
+  {
+    uint from_len= (m_field_metadata[col] >> 8U) & 0x00ff;
+    uint from_bit_len= m_field_metadata[col] & 0x00ff;
+    length= from_len + ((from_bit_len > 0) ? 1 : 0);
     break;
-
-  case MYSQL_TYPE_ENUM:
-  case MYSQL_TYPE_SET:
-  case MYSQL_TYPE_VAR_STRING:
+  }
   case MYSQL_TYPE_VARCHAR:
-    length= ~(uint32) 0;                               // NYI
+    length= m_field_metadata[col] > 255 ? 2 : 1; // c&p of Field_varstring::data_length()
+    length+= length == 1 ? (uint32) *master_data : uint2korr(master_data);
     break;
-
   case MYSQL_TYPE_TINY_BLOB:
   case MYSQL_TYPE_MEDIUM_BLOB:
   case MYSQL_TYPE_LONG_BLOB:
   case MYSQL_TYPE_BLOB:
   case MYSQL_TYPE_GEOMETRY:
-    length= ~(uint32) 0;                               // NYI
+  {
+    Field_blob fb(m_field_metadata[col]);
+    length= fb.get_packed_size(master_data);
     break;
   }
-
+  default:
+    length= -1;
+  }
   return length;
 }
 
-/*********************************************************************
- *                   table_def member definitions                    *
- *********************************************************************/
-
 /*
   Is the definition compatible with a table?
 
@@ -121,23 +133,6 @@ table_def::compatible_with(RELAY_LOG_INFO const *rli_arg, TABLE *table)
 
   TABLE_SHARE const *const tsh= table->s;
 
-  /*
-    To get proper error reporting for all columns of the table, we
-    both check the width and iterate over all columns.
-  */
-  if (tsh->fields < size())
-  {
-    DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
-    error= 1;
-    char buf[256];
-    my_snprintf(buf, sizeof(buf), "Table width mismatch - "
-                "received %u columns, %s.%s has %u columns",
-                (uint) size(), tsh->db.str, tsh->table_name.str,
-                tsh->fields);
-    rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
-                ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
-  }
-
   for (uint col= 0 ; col < cols_to_check ; ++col)
   {
     if (table->field[col]->type() != type(col))
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index 79e69aecaeb..c91113c9e2a 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -25,8 +25,6 @@
 struct st_relay_log_info;
 typedef st_relay_log_info RELAY_LOG_INFO;
 
-uint32
-field_length_from_packed(enum_field_types field_type, uchar const *data);
 
 /**
   A table definition from the master.
@@ -57,19 +55,96 @@ public:
 
     @param types Array of types
     @param size  Number of elements in array 'types'
+    @param field_metadata Array of extra information about fields
+    @param metadata_size Size of the field_metadata array
+    @param null_bitmap The bitmap of fields that can be null
    */
-  table_def(field_type *types, ulong size)
-    : m_size(size), m_type(new unsigned char [size])
+  table_def(field_type *types, ulong size, uchar *field_metadata, 
+      int metadata_size, uchar *null_bitmap)
+    : m_size(size), m_type(0),
+      m_field_metadata(0), m_null_bits(0), m_memory(NULL)
   {
+    m_memory= (uchar *)my_multi_malloc(MYF(MY_WME),
+                              &m_type, size,
+                              &m_field_metadata, size * sizeof(short),
+                              &m_null_bits, (m_size + 7) / 8,
+                              NULL);
     if (m_type)
       memcpy(m_type, types, size);
     else
       m_size= 0;
+    /*
+      Extract the data from the table map into the field metadata array
+      iff there is field metadata. The variable metadata_size will be
+      0 if we are replicating from an older version server since no field
+      metadata was written to the table map. This can also happen if 
+      there were no fields in the master that needed extra metadata.
+    */
+    if (m_size && metadata_size)
+    { 
+      int index= 0;
+      for (unsigned int i= 0; i < m_size; i++)
+      {
+        switch (m_type[i]) {
+        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:
+        {
+          /*
+            These types store a single byte.
+          */
+          m_field_metadata[i]= (uchar)field_metadata[index];
+          index++;
+          break;
+        }
+        case MYSQL_TYPE_SET:
+        case MYSQL_TYPE_ENUM:
+        case MYSQL_TYPE_STRING:
+        {
+          short int x= field_metadata[index++] << 8U; // real_type
+          x = x + field_metadata[index++];            // pack or field length
+          m_field_metadata[i]= x;
+          break;
+        }
+        case MYSQL_TYPE_BIT:
+        {
+          short int x= field_metadata[index++]; 
+          x = x + (field_metadata[index++] << 8U);
+          m_field_metadata[i]= x;
+          break;
+        }
+        case MYSQL_TYPE_VARCHAR:
+        {
+          /*
+            These types store two bytes.
+          */
+          char *ptr= (char *)&field_metadata[index];
+          m_field_metadata[i]= sint2korr(ptr);
+          index= index + sizeof(short int);
+          break;
+        }
+        case MYSQL_TYPE_NEWDECIMAL:
+        {
+          short int x= field_metadata[index++] << 8U; // precision
+          x = x + field_metadata[index++];            // decimals
+          m_field_metadata[i]= x;
+          break;
+        }
+        default:
+          m_field_metadata[i]= 0;
+          break;
+        }
+      }
+    }
+    if (m_size && null_bitmap)
+       memcpy(m_null_bits, null_bitmap, (m_size + 7) / 8);
   }
 
   ~table_def() {
-    if (m_type)
-      delete [] m_type;
+    my_free(m_memory, MYF(0));
 #ifndef DBUG_OFF
     m_type= 0;
     m_size= 0;
@@ -99,6 +174,48 @@ public:
     return m_type[index];
   }
 
+
+  /*
+    This function allows callers to get the extra field data from the
+    table map for a given field. If there is no metadata for that field
+    or there is no extra metadata at all, the function returns 0.
+
+    The function returns the value for the field metadata for column at 
+    position indicated by index. As mentioned, if the field was a type 
+    that stores field metadata, that value is returned else zero (0) is 
+    returned. This method is used in the unpack() methods of the 
+    corresponding fields to properly extract the data from the binary log 
+    in the event that the master's field is smaller than the slave.
+  */
+  uint16 field_metadata(uint index) const
+  {
+    DBUG_ASSERT(index < m_size);
+    if (m_field_metadata)
+      return m_field_metadata[index];
+    else
+      return 0;
+  }
+
+  /*
+    This function returns whether the field on the master can be null.
+    This value is derived from field->maybe_null().
+  */
+  my_bool maybe_null(uint index) const
+  {
+    DBUG_ASSERT(index < m_size);
+    return ((m_null_bits[(index / 8)] & 
+            (1 << (index % 8))) == (1 << (index %8)));
+  }
+
+  /*
+    This function returns the field size in raw bytes based on the type
+    and the encoded field data from the master's raw data. This method can 
+    be used for situations where the slave needs to skip a column (e.g., 
+    WL#3915) or needs to advance the pointer for the fields in the raw 
+    data from the master to a specific column.
+  */
+  uint32 calc_field_size(uint col, uchar *master_data);
+
   /**
     Decide if the table definition is compatible with a table.
 
@@ -121,6 +238,9 @@ public:
 private:
   ulong m_size;           // Number of elements in the types array
   field_type *m_type;                     // Array of type descriptors
+  short int *m_field_metadata;
+  uchar *m_null_bits;
+  uchar *m_memory;
 };
 
 /**