BUG#37426: RBR breaks for CHAR() UTF-8 fields > 85 chars

      
In order to handle CHAR() fields, 8 bits were reserved for
the size of the CHAR field. However, instead of denoting the
number of characters in the field, field_length was used which
denotes the number of bytes in the field.

Since UTF-8 fields can have three bytes per character (and
has been extended to have four bytes per character in 6.0),
an extra two bits have been encoded in the field metadata
work for fields of type Field_string (i.e., CHAR fields).

Since the metadata word is filled, the extra bits have been
encoded in the upper 4 bits of the real type (the most 
significant byte of the metadata word) by computing the
bitwise xor of the extra two bits. Since the upper 4 bits
of the real type always is 1111 for Field_string, this 
means that for fields of length <256, the encoding is
identical to the encoding used in pre-5.1.26 servers, but
for lengths of 256 or more, an unrecognized type is formed,
causing an old slave (that does not handle lengths of 256
or more) to stop.

6 files changed, 124 insertions, 29 deletions

diff --git a/sql/field.cc b/sql/field.cc
index 9bf6df55b30..21e26b518e8 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -27,6 +27,8 @@
 
 #include "mysql_priv.h"
 #include "sql_select.h"
+#include "rpl_rli.h"                            // Pull in Relay_log_info
+#include "slave.h"                              // Pull in rpl_master_has_bug()
 #include <m_ctype.h>
 #include <errno.h>
 #ifdef HAVE_FCONVERT
@@ -1375,7 +1377,8 @@ bool Field::send_binary(Protocol *protocol)
    @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)
+int Field::compatible_field_size(uint field_metadata,
+                                 const Relay_log_info *rli_arg __attribute__((unused)))
 {
   uint const source_size= pack_length_from_metadata(field_metadata);
   uint const destination_size= row_pack_length();
@@ -2837,7 +2840,8 @@ uint Field_new_decimal::pack_length_from_metadata(uint 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 Field_new_decimal::compatible_field_size(uint field_metadata,
+                                             const Relay_log_info * __attribute__((unused)))
 {
   int compatible= 0;
   uint const source_precision= (field_metadata >> 8U) & 0x00ff;
@@ -4037,7 +4041,6 @@ Field_real::pack(uchar *to, const uchar *from,
 {
   DBUG_ENTER("Field_real::pack");
   DBUG_ASSERT(max_length >= pack_length());
-  DBUG_PRINT("debug", ("pack_length(): %u", pack_length()));
 #ifdef WORDS_BIGENDIAN
   if (low_byte_first != table->s->db_low_byte_first)
   {
@@ -4056,7 +4059,6 @@ Field_real::unpack(uchar *to, const uchar *from,
                    uint param_data, bool low_byte_first)
 {
   DBUG_ENTER("Field_real::unpack");
-  DBUG_PRINT("debug", ("pack_length(): %u", pack_length()));
 #ifdef WORDS_BIGENDIAN
   if (low_byte_first != table->s->db_low_byte_first)
   {
@@ -6638,6 +6640,36 @@ my_decimal *Field_string::val_decimal(my_decimal *decimal_value)
 }
 
 
+struct Check_field_param {
+  Field *field;
+};
+
+static bool
+check_field_for_37426(const void *param_arg)
+{
+  Check_field_param *param= (Check_field_param*) param_arg;
+  DBUG_ASSERT(param->field->real_type() == MYSQL_TYPE_STRING);
+  DBUG_PRINT("debug", ("Field %s - type: %d, size: %d",
+                       param->field->field_name,
+                       param->field->real_type(),
+                       param->field->row_pack_length()));
+  return param->field->row_pack_length() > 255;
+}
+
+
+int Field_string::compatible_field_size(uint field_metadata,
+                                        const Relay_log_info *rli_arg)
+{
+#ifdef HAVE_REPLICATION
+  const Check_field_param check_param = { this };
+  if (rpl_master_has_bug(rli_arg, 37426, TRUE,
+                         check_field_for_37426, &check_param))
+    return FALSE;                        // Not compatible field sizes
+#endif
+  return Field::compatible_field_size(field_metadata, rli_arg);
+}
+
+
 int Field_string::cmp(const uchar *a_ptr, const uchar *b_ptr)
 {
   uint a_len, b_len;
@@ -6724,6 +6756,9 @@ uchar *Field_string::pack(uchar *to, const uchar *from,
    @c param_data argument contains the result of field->real_type() from
    the master.
 
+   @note For information about how the length is packed, see @c
+   Field_string::do_save_field_metadata
+
    @param   to         Destination of the data
    @param   from       Source of the data
    @param   param_data Real type (upper) and length (lower) values
@@ -6736,10 +6771,24 @@ Field_string::unpack(uchar *to,
                      uint param_data,
                      bool low_byte_first __attribute__((unused)))
 {
-  uint from_length=
-    param_data ? min(param_data & 0x00ff, field_length) : field_length;
-  uint length;
+  uint from_length, length;
 
+  /*
+    Compute the declared length of the field on the master. This is
+    used to decide if one or two bytes should be read as length.
+   */
+  if (param_data)
+    from_length= (((param_data >> 4) & 0x300) ^ 0x300) + (param_data & 0x00ff);
+  else
+    from_length= field_length;
+
+  DBUG_PRINT("debug",
+             ("param_data: 0x%x, field_length: %u, from_length: %u",
+              param_data, field_length, from_length));
+  /*
+    Compute the actual length of the data by reading one or two bits
+    (depending on the declared field length on the master).
+   */
   if (from_length > 255)
   {
     length= uint2korr(from);
@@ -6762,14 +6811,37 @@ Field_string::unpack(uchar *to,
    second byte of the field metadata array at index of *metadata_ptr and
    *(metadata_ptr + 1).
 
+   @note In order to be able to handle lengths exceeding 255 and be
+   backwards-compatible with pre-5.1.26 servers, an extra two bits of
+   the length has been added to the metadata in such a way that if
+   they are set, a new unrecognized type is generated.  This will
+   cause pre-5.1-26 servers to stop due to a field type mismatch,
+   while new servers will be able to extract the extra bits. If the
+   length is <256, there will be no difference and both a new and an
+   old server will be able to handle it.
+
+   @note The extra two bits are added to bits 13 and 14 of the
+   parameter data (with 1 being the least siginficant bit and 16 the
+   most significant bit of the word) by xoring the extra length bits
+   with the real type.  Since all allowable types have 0xF as most
+   significant bits of the metadata word, lengths <256 will not affect
+   the real type at all, while all other values will result in a
+   non-existant type in the range 17-244.
+
+   @see Field_string::unpack
+
    @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;
+  DBUG_ASSERT(field_length < 1024);
+  DBUG_ASSERT((real_type() & 0xF0) == 0xF0);
+  DBUG_PRINT("debug", ("field_length: %u, real_type: %u",
+                       field_length, real_type()));
+  *metadata_ptr= (real_type() ^ ((field_length & 0x300) >> 4));
+  *(metadata_ptr + 1)= field_length & 0xFF;
   return 2;
 }
 
@@ -9118,7 +9190,8 @@ uint Field_bit::pack_length_from_metadata(uint 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 Field_bit::compatible_field_size(uint field_metadata,
+                                     const Relay_log_info * __attribute__((unused)))
 {
   int compatible= 0;
   uint const source_size= pack_length_from_metadata(field_metadata);
diff --git a/sql/field.h b/sql/field.h
index 479a71dfee5..6c6cab1e4f0 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -30,6 +30,8 @@ const uint32 max_field_size= (uint32) 4294967295U;
 class Send_field;
 class Protocol;
 class Create_field;
+class Relay_log_info;
+
 struct st_cache_field;
 int field_conv(Field *to,Field *from);
 
@@ -162,7 +164,8 @@ 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 int compatible_field_size(uint field_metadata,
+                                    const Relay_log_info *);
   virtual uint pack_length_from_metadata(uint field_metadata)
   { return field_metadata; }
   /*
@@ -716,7 +719,8 @@ public:
   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);
+  int compatible_field_size(uint field_metadata,
+                            const Relay_log_info *rli);
   uint is_equal(Create_field *new_field);
   virtual const uchar *unpack(uchar* to, const uchar *from,
                               uint param_data, bool low_byte_first);
@@ -1425,7 +1429,14 @@ public:
   virtual const uchar *unpack(uchar* to, const uchar *from,
                               uint param_data, bool low_byte_first);
   uint pack_length_from_metadata(uint field_metadata)
-  { return (field_metadata & 0x00ff); }
+  {
+    DBUG_PRINT("debug", ("field_metadata: 0x%04x", field_metadata));
+    if (field_metadata == 0)
+      return row_pack_length();
+    return (((field_metadata >> 4) & 0x300) ^ 0x300) + (field_metadata & 0x00ff);
+  }
+  int compatible_field_size(uint field_metadata,
+                            const Relay_log_info *rli);
   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);
@@ -1878,7 +1889,8 @@ public:
   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);
+  int compatible_field_size(uint field_metadata,
+                            const Relay_log_info *rli);
   void sql_type(String &str) const;
   virtual uchar *pack(uchar *to, const uchar *from,
                       uint max_length, bool low_byte_first);
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index 4f4083d9b8f..e34f8561051 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -188,7 +188,8 @@ table_def::compatible_with(Relay_log_info const *rli_arg, TABLE *table)
 
   for (uint col= 0 ; col < cols_to_check ; ++col)
   {
-    if (table->field[col]->type() != type(col))
+    Field *const field= table->field[col];
+    if (field->type() != type(col))
     {
       DBUG_ASSERT(col < size() && col < tsh->fields);
       DBUG_ASSERT(tsh->db.str && tsh->table_name.str);
@@ -197,15 +198,15 @@ table_def::compatible_with(Relay_log_info const *rli_arg, TABLE *table)
       my_snprintf(buf, sizeof(buf), "Column %d type mismatch - "
                   "received type %d, %s.%s has type %d",
                   col, type(col), tsh->db.str, tsh->table_name.str,
-                  table->field[col]->type());
+                  field->type());
       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)))
+    if (!error &&
+        !field->compatible_field_size(field_metadata(col), rli_arg))
     {
       error= 1;
       char buf[256];
@@ -213,10 +214,9 @@ table_def::compatible_with(Relay_log_info const *rli_arg, TABLE *table)
                   "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());
+                  field->pack_length_from_metadata(m_field_metadata[col]),
+                  tsh->db.str, tsh->table_name.str,
+                  field->row_pack_length());
       rli->report(ERROR_LEVEL, ER_BINLOG_ROW_WRONG_TABLE_DEF,
                   ER(ER_BINLOG_ROW_WRONG_TABLE_DEF), buf);
     }
diff --git a/sql/slave.cc b/sql/slave.cc
index a6d7758c8de..0040b69f8de 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -4057,9 +4057,17 @@ end:
    @param rli Relay_log_info which tells the master's version
    @param bug_id Number of the bug as found in bugs.mysql.com
    @param report bool report error message, default TRUE
+
+   @param pred Predicate function that will be called with @c param to
+   check for the bug. If the function return @c true, the bug is present,
+   otherwise, it is not.
+
+   @param param  State passed to @c pred function.
+
    @return TRUE if master has the bug, FALSE if it does not.
 */
-bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id, bool report)
+bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
+                        bool (*pred)(const void *), const void *param)
 {
   struct st_version_range_for_one_bug {
     uint        bug_id;
@@ -4072,6 +4080,7 @@ bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id, bool report)
     {24432, { 5, 1, 12 }, { 5, 1, 17 } },
     {33029, { 5, 0,  0 }, { 5, 0, 58 } },
     {33029, { 5, 1,  0 }, { 5, 1, 12 } },
+    {37426, { 5, 1,  0 }, { 5, 1, 26 } },
   };
   const uchar *master_ver=
     rli->relay_log.description_event_for_exec->server_version_split;
@@ -4085,11 +4094,11 @@ bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id, bool report)
       *fixed_in= versions_for_all_bugs[i].fixed_in;
     if ((versions_for_all_bugs[i].bug_id == bug_id) &&
         (memcmp(introduced_in, master_ver, 3) <= 0) &&
-        (memcmp(fixed_in,      master_ver, 3) >  0))
+        (memcmp(fixed_in,      master_ver, 3) >  0) &&
+        (pred == NULL || (*pred)(param)))
     {
       if (!report)
 	return TRUE;
-      
       // a short message for SHOW SLAVE STATUS (message length constraints)
       my_printf_error(ER_UNKNOWN_ERROR, "master may suffer from"
                       " http://bugs.mysql.com/bug.php?id=%u"
@@ -4137,7 +4146,7 @@ bool rpl_master_erroneous_autoinc(THD *thd)
   {
     Relay_log_info *rli= &active_mi->rli;
     DBUG_EXECUTE_IF("simulate_bug33029", return TRUE;);
-    return rpl_master_has_bug(rli, 33029, FALSE);
+    return rpl_master_has_bug(rli, 33029, FALSE, NULL, NULL);
   }
   return FALSE;
 }
diff --git a/sql/slave.h b/sql/slave.h
index 80d267e5b27..dc2d668c97b 100644
--- a/sql/slave.h
+++ b/sql/slave.h
@@ -165,7 +165,8 @@ int fetch_master_table(THD* thd, const char* db_name, const char* table_name,
 
 bool show_master_info(THD* thd, Master_info* mi);
 bool show_binlog_info(THD* thd);
-bool rpl_master_has_bug(Relay_log_info *rli, uint bug_id, bool report=TRUE);
+bool rpl_master_has_bug(const Relay_log_info *rli, uint bug_id, bool report,
+                        bool (*pred)(const void *), const void *param);
 bool rpl_master_erroneous_autoinc(THD* thd);
 
 const char *print_slave_db_safe(const char *db);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index d1e5c26912d..c51c47ae431 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -695,7 +695,7 @@ bool mysql_insert(THD *thd,TABLE_LIST *table_list,
   if (thd->slave_thread &&
       (info.handle_duplicates == DUP_UPDATE) &&
       (table->next_number_field != NULL) &&
-      rpl_master_has_bug(&active_mi->rli, 24432))
+      rpl_master_has_bug(&active_mi->rli, 24432, TRUE, NULL, NULL))
     goto abort;
 #endif
 
@@ -2967,7 +2967,7 @@ select_insert::prepare(List<Item> &values, SELECT_LEX_UNIT *u)
   if (thd->slave_thread &&
       (info.handle_duplicates == DUP_UPDATE) &&
       (table->next_number_field != NULL) &&
-      rpl_master_has_bug(&active_mi->rli, 24432))
+      rpl_master_has_bug(&active_mi->rli, 24432, TRUE, NULL, NULL))
     DBUG_RETURN(1);
 #endif