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.


mysql-test/extra/rpl_tests/rpl_row_basic.test:
  Adding test cases for replicating UTF-8 fields of lengths
  of 256 or more (bytes).
mysql-test/suite/binlog/r/binlog_base64_flag.result:
  Result file change.
mysql-test/suite/binlog/t/binlog_base64_flag.test:
  Adding tests to trigger check that an error is generated when replicating from a
  5.1.25 server for tables with a CHAR(128) but not when replicating a table with a
  CHAR(63). Although the bug indicates that the limit is 83, we elected to use CHAR(63)
  since 6.0 uses 4-byte UTF-8, and anything exceeding 63 would then cause the test to fail
  when the patch is merged to 6.0.
mysql-test/suite/bugs/combinations:
  Adding combinations file to run all bug reports in all binlog modes (where
  applicable).
mysql-test/suite/bugs/r/rpl_bug37426.result:
  Result file change.
mysql-test/suite/bugs/t/rpl_bug37426.test:
  Added test for reported bug.
mysql-test/suite/rpl/r/rpl_row_basic_2myisam.result:
  Result file change.
mysql-test/suite/rpl/r/rpl_row_basic_3innodb.result:
  Result file change.
sql/field.cc:
  Encoding an extra two bits in the most significant nibble (4 bits)
  of the metadata word. Adding assertions to ensure that no attempt
  is made to use lengths longer than supported.
  
  Extending compatible_field_size() function with an extra parameter
  holding a Relay_log_instace for error reporting.
  
  Field_string::compatible_field_size() now reports an error if field
  size for a CHAR is >255.
sql/field.h:
  Field length is now computed from most significant 4 bits
  of metadata word, or is equal to the row pack length if
  there is no metadata.
  
  Extending compatible_field_size() function with an extra parameter
  holding a Relay_log_instace for error reporting.
sql/rpl_utility.cc:
  Adding relay log parameter to compatible_field_size().
  
  Minor refactoring to eliminate duplicate code.
sql/slave.cc:
  Extending rpl_master_has_bug() with a single-argument predicate function and
  a parameter to the predicate function. The predicate function can be used to
  test for extra conditions for the bug before writing an error message.
sql/slave.h:
  Extending rpl_master_has_bug() with a single-argument predicate function and
  a parameter to the predicate function. The predicate function can be used to
  test for extra conditions for the bug before writing an error message.
  
  Also removing gratuitous default argument.
sql/sql_insert.cc:
  Changing calls to rpl_master_has_bug() to adapt to changed signature.

14 files changed, 517 insertions, 35 deletions

diff --git a/mysql-test/extra/rpl_tests/rpl_row_basic.test b/mysql-test/extra/rpl_tests/rpl_row_basic.test
index 48ddbaf244a..59f049e2bfa 100644
--- a/mysql-test/extra/rpl_tests/rpl_row_basic.test
+++ b/mysql-test/extra/rpl_tests/rpl_row_basic.test
@@ -259,7 +259,7 @@ DELETE FROM t1;
 query_vertical SELECT COUNT(*) FROM t1 ORDER BY c1,c2;
 sync_slave_with_master;
 set @@global.slave_exec_mode= default;
-let $last_error = query_get_value("SHOW SLAVE STATUS", Last_SQL_Errno, 1);
+let $last_error = query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
 disable_query_log;
 eval SELECT "$last_error" AS Last_SQL_Error;
 enable_query_log;
@@ -272,3 +272,150 @@ query_vertical SELECT COUNT(*) FROM t1 ORDER BY c1,c2;
 connection master;
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7,t8;
 sync_slave_with_master;
+
+#
+# BUG#37426: RBR breaks for CHAR() UTF8 fields > 85 chars
+#
+
+# We have 4 combinations to test with respect to the field length
+# (i.e., the number of bytes) of the CHAR fields:
+#
+# 1. Replicating from CHAR<256 to CHAR<256 
+# 2. Replicating from CHAR<256 to CHAR>255
+# 3. Replicating from CHAR>255 to CHAR<256
+# 4. Replicating from CHAR>255 to CHAR>255
+
+# We also make a special case of using the max size of a field on the
+# master, i.e. CHAR(255) in UTF-8, giving another three cases.
+#
+# 5. Replicating UTF-8 CHAR(255) to CHAR(<256)
+# 6. Replicating UTF-8 CHAR(255) to CHAR(>255)
+# 7. Replicating UTF-8 CHAR(255) to CHAR(255) UTF-8
+
+connection master;
+CREATE TABLE t1 (i INT NOT NULL,
+                 c CHAR(16) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+
+CREATE TABLE t2 (i INT NOT NULL,
+                 c CHAR(16) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+
+sync_slave_with_master;
+ALTER TABLE t2 MODIFY c CHAR(128) CHARACTER SET utf8 NOT NULL;
+
+connection master;
+CREATE TABLE t3 (i INT NOT NULL,
+                 c CHAR(128) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+sync_slave_with_master;
+ALTER TABLE t3 MODIFY c CHAR(16) CHARACTER SET utf8 NOT NULL;
+
+connection master;
+CREATE TABLE t4 (i INT NOT NULL,
+                 c CHAR(128) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+
+CREATE TABLE t5 (i INT NOT NULL,
+                 c CHAR(255) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+sync_slave_with_master;
+ALTER TABLE t5 MODIFY c CHAR(16) CHARACTER SET utf8 NOT NULL;
+
+connection master;
+CREATE TABLE t6 (i INT NOT NULL,
+                 c CHAR(255) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+sync_slave_with_master;
+ALTER TABLE t6 MODIFY c CHAR(128) CHARACTER SET utf8 NOT NULL;
+
+connection master;
+CREATE TABLE t7 (i INT NOT NULL,
+                 c CHAR(255) CHARACTER SET utf8 NOT NULL,
+                 j INT NOT NULL);
+
+--echo [expecting slave to replicate correctly]
+connection master;
+INSERT INTO t1 VALUES (1, "", 1);
+INSERT INTO t1 VALUES (2, repeat(_utf8'a', 16), 2);
+
+let $diff_table_1=master:test.t1;
+let $diff_table_2=slave:test.t1;
+source include/diff_tables.inc;
+
+--echo [expecting slave to replicate correctly]
+connection master;
+INSERT INTO t2 VALUES (1, "", 1);
+INSERT INTO t2 VALUES (2, repeat(_utf8'a', 16), 2);
+
+let $diff_table_1=master:test.t2;
+let $diff_table_2=slave:test.t2;
+source include/diff_tables.inc;
+
+--echo [expecting slave to stop]
+connection master;
+INSERT INTO t3 VALUES (1, "", 1);
+INSERT INTO t3 VALUES (2, repeat(_utf8'a', 128), 2);
+
+connection slave;
+source include/wait_for_slave_sql_to_stop.inc;
+let $last_error = query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
+disable_query_log;
+eval SELECT "$last_error" AS Last_SQL_Error;
+enable_query_log;
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+source include/wait_for_slave_to_start.inc;
+
+--echo [expecting slave to replicate correctly]
+connection master;
+INSERT INTO t4 VALUES (1, "", 1);
+INSERT INTO t4 VALUES (2, repeat(_utf8'a', 128), 2);
+
+let $diff_table_1=master:test.t4;
+let $diff_table_2=slave:test.t4;
+source include/diff_tables.inc;
+
+--echo [expecting slave to stop]
+connection master;
+INSERT INTO t5 VALUES (1, "", 1);
+INSERT INTO t5 VALUES (2, repeat(_utf8'a', 255), 2);
+
+connection slave;
+source include/wait_for_slave_sql_to_stop.inc;
+let $last_error = query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
+disable_query_log;
+eval SELECT "$last_error" AS Last_SQL_Error;
+enable_query_log;
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+source include/wait_for_slave_to_start.inc;
+
+--echo [expecting slave to stop]
+connection master;
+INSERT INTO t6 VALUES (1, "", 1);
+INSERT INTO t6 VALUES (2, repeat(_utf8'a', 255), 2);
+
+connection slave;
+source include/wait_for_slave_sql_to_stop.inc;
+let $last_error = query_get_value("SHOW SLAVE STATUS", Last_SQL_Error, 1);
+disable_query_log;
+eval SELECT "$last_error" AS Last_SQL_Error;
+enable_query_log;
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+source include/wait_for_slave_to_start.inc;
+
+--echo [expecting slave to replicate correctly]
+connection master;
+INSERT INTO t7 VALUES (1, "", 1);
+INSERT INTO t7 VALUES (2, repeat(_utf8'a', 255), 2);
+
+let $diff_table_1=master:test.t7;
+let $diff_table_2=slave:test.t7;
+source include/diff_tables.inc;
+
+connection master;
+drop table t1, t2, t3, t4, t5, t6, t7;
+sync_slave_with_master;
+
diff --git a/mysql-test/suite/binlog/r/binlog_base64_flag.result b/mysql-test/suite/binlog/r/binlog_base64_flag.result
index 72a16010031..fbc2b8344ee 100644
--- a/mysql-test/suite/binlog/r/binlog_base64_flag.result
+++ b/mysql-test/suite/binlog/r/binlog_base64_flag.result
@@ -66,4 +66,28 @@ a
 1
 1
 3
-drop table t1;
+CREATE TABLE char128_utf8 ( 
+i1 INT NOT NULL, 
+c CHAR(128) CHARACTER SET utf8 NOT NULL, 
+i2 INT NOT NULL);
+CREATE TABLE char63_utf8 (
+i1 INT NOT NULL,
+c CHAR(63) CHARACTER SET utf8 NOT NULL,
+i2 INT NOT NULL);
+BINLOG '
+MuNkSA8BAAAAZgAAAGoAAAAAAAQANS4xLjI1LXJjLWRlYnVnLWxvZwAAAAAAAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAy42RIEzgNAAgAEgAEBAQEEgAAUwAEGggAAAAICAgC
+';
+BINLOG '
+3u9kSBMBAAAANgAAAJYBAAAAABAAAAAAAAAABHRlc3QAC2NoYXI2M191dGY4AAMD/gMC/r0A
+3u9kSBcBAAAAKgAAAMABAAAQABAAAAAAAAEAA//4AQAAAAMxMjMBAAAA
+';
+SELECT * FROM char63_utf8;
+i1	c	i2
+1	123	1
+BINLOG '
+iONkSBMBAAAANwAAAJkBAAAAABAAAAAAAAAABHRlc3QADGNoYXIxMjhfdXRmOAADA/4DAv6AAA==
+iONkSBcBAAAAKwAAAMQBAAAQABAAAAAAAAEAA//4AQAAAAMAMTIzAQAAAA==
+';
+ERROR HY000: master may suffer from http://bugs.mysql.com/bug.php?id=37426 so slave stops; check error log on slave for more info
+drop table t1, char63_utf8, char128_utf8;
diff --git a/mysql-test/suite/binlog/t/binlog_base64_flag.test b/mysql-test/suite/binlog/t/binlog_base64_flag.test
index abf35a6bfd8..e6271ec6ccc 100644
--- a/mysql-test/suite/binlog/t/binlog_base64_flag.test
+++ b/mysql-test/suite/binlog/t/binlog_base64_flag.test
@@ -104,6 +104,49 @@ Dl1YRxcBAAAAIgAAAFYBAAAQABAAAAAAAAEAAf/+BQAAAA==
 # the above line should fail and 5 should not be in the binlog.
 select * from t1;
 
+# Test that BUG#37426 is triggered.
 
-# clean up
-drop table t1;
+CREATE TABLE char128_utf8 ( 
+    i1 INT NOT NULL, 
+    c CHAR(128) CHARACTER SET utf8 NOT NULL, 
+    i2 INT NOT NULL);
+CREATE TABLE char63_utf8 (
+      i1 INT NOT NULL,
+      c CHAR(63) CHARACTER SET utf8 NOT NULL,
+      i2 INT NOT NULL);
+
+#
+# This is the format description log event
+#
+
+BINLOG '
+MuNkSA8BAAAAZgAAAGoAAAAAAAQANS4xLjI1LXJjLWRlYnVnLWxvZwAAAAAAAAAAAAAAAAAAAAAA
+AAAAAAAAAAAAAAAAAAAy42RIEzgNAAgAEgAEBAQEEgAAUwAEGggAAAAICAgC
+';
+
+# ... this event corresponding to
+#
+#    INSERT INTO char63_utf8 VALUES ( 1, "123", 1 )
+#
+# The binlog event below shall not trigger the bug check
+
+BINLOG '
+3u9kSBMBAAAANgAAAJYBAAAAABAAAAAAAAAABHRlc3QAC2NoYXI2M191dGY4AAMD/gMC/r0A
+3u9kSBcBAAAAKgAAAMABAAAQABAAAAAAAAEAA//4AQAAAAMxMjMBAAAA
+';
+SELECT * FROM char63_utf8;
+
+# ... and this is an event corresponding to
+#
+#    INSERT INTO char128_utf8 VALUES ( 1, "123", 1 )
+#
+# The binlog event below shall trigger the bug check and produce an error
+#
+
+error ER_UNKNOWN_ERROR;
+BINLOG '
+iONkSBMBAAAANwAAAJkBAAAAABAAAAAAAAAABHRlc3QADGNoYXIxMjhfdXRmOAADA/4DAv6AAA==
+iONkSBcBAAAAKwAAAMQBAAAQABAAAAAAAAEAA//4AQAAAAMAMTIzAQAAAA==
+';
+
+drop table t1, char63_utf8, char128_utf8;
diff --git a/mysql-test/suite/bugs/combinations b/mysql-test/suite/bugs/combinations
new file mode 100644
index 00000000000..ea25611a5d4
--- /dev/null
+++ b/mysql-test/suite/bugs/combinations
@@ -0,0 +1,8 @@
+[row]
+--binlog-format=row
+
+[stmt]
+--binlog-format=statement
+
+[mix]
+--binlog-format=mixed
diff --git a/mysql-test/suite/bugs/r/rpl_bug37426.result b/mysql-test/suite/bugs/r/rpl_bug37426.result
new file mode 100644
index 00000000000..24dfd27ca01
--- /dev/null
+++ b/mysql-test/suite/bugs/r/rpl_bug37426.result
@@ -0,0 +1,17 @@
+stop slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+reset master;
+reset slave;
+drop table if exists t1,t2,t3,t4,t5,t6,t7,t8,t9;
+start slave;
+CREATE TABLE char128_utf8 ( 
+i1 INT NOT NULL, 
+c CHAR(128) CHARACTER SET utf8 NOT NULL, 
+i2 INT NOT NULL);
+INSERT INTO char128_utf8 VALUES ( 1, "123", 1 );
+SELECT * FROM char128_utf8;
+i1	c	i2
+1	123	1
+SELECT * FROM char128_utf8;
+i1	c	i2
+1	123	1
diff --git a/mysql-test/suite/bugs/t/rpl_bug37426.test b/mysql-test/suite/bugs/t/rpl_bug37426.test
new file mode 100644
index 00000000000..d2bad0fa0c1
--- /dev/null
+++ b/mysql-test/suite/bugs/t/rpl_bug37426.test
@@ -0,0 +1,22 @@
+#############################################################
+# Author: Mats Kindahl <mats@mysql.com>
+# Date:   2008-06-18
+# Purpose: Test for BUG#37426
+#   RBR breaks for CHAR() UTF8 fields > 85 chars
+#############################################################
+
+source include/master-slave.inc;
+source include/have_binlog_format_row.inc;
+
+connection master;
+CREATE TABLE char128_utf8 ( 
+    i1 INT NOT NULL, 
+    c CHAR(128) CHARACTER SET utf8 NOT NULL, 
+    i2 INT NOT NULL);
+
+INSERT INTO char128_utf8 VALUES ( 1, "123", 1 );
+
+SELECT * FROM char128_utf8;
+sync_slave_with_master;
+
+SELECT * FROM char128_utf8;
diff --git a/mysql-test/suite/rpl/r/rpl_row_basic_2myisam.result b/mysql-test/suite/rpl/r/rpl_row_basic_2myisam.result
index 22284a26412..3977e721212 100644
--- a/mysql-test/suite/rpl/r/rpl_row_basic_2myisam.result
+++ b/mysql-test/suite/rpl/r/rpl_row_basic_2myisam.result
@@ -437,7 +437,70 @@ SELECT COUNT(*) FROM t1 ORDER BY c1,c2;
 COUNT(*)	0
 set @@global.slave_exec_mode= default;
 Last_SQL_Error
-0
+
 SELECT COUNT(*) FROM t1 ORDER BY c1,c2;
 COUNT(*)	0
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7,t8;
+CREATE TABLE t1 (i INT NOT NULL,
+c CHAR(16) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+CREATE TABLE t2 (i INT NOT NULL,
+c CHAR(16) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t2 MODIFY c CHAR(128) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t3 (i INT NOT NULL,
+c CHAR(128) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t3 MODIFY c CHAR(16) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t4 (i INT NOT NULL,
+c CHAR(128) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+CREATE TABLE t5 (i INT NOT NULL,
+c CHAR(255) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t5 MODIFY c CHAR(16) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t6 (i INT NOT NULL,
+c CHAR(255) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t6 MODIFY c CHAR(128) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t7 (i INT NOT NULL,
+c CHAR(255) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+[expecting slave to replicate correctly]
+INSERT INTO t1 VALUES (1, "", 1);
+INSERT INTO t1 VALUES (2, repeat(_utf8'a', 16), 2);
+Comparing tables master:test.t1 and slave:test.t1
+[expecting slave to replicate correctly]
+INSERT INTO t2 VALUES (1, "", 1);
+INSERT INTO t2 VALUES (2, repeat(_utf8'a', 16), 2);
+Comparing tables master:test.t2 and slave:test.t2
+[expecting slave to stop]
+INSERT INTO t3 VALUES (1, "", 1);
+INSERT INTO t3 VALUES (2, repeat(_utf8'a', 128), 2);
+Last_SQL_Error
+Table definition on master and slave does not match: Column 1 size mismatch - master has size 384, test.t3 on slave has size 49. Master's column size should be <= the slave's column size.
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+[expecting slave to replicate correctly]
+INSERT INTO t4 VALUES (1, "", 1);
+INSERT INTO t4 VALUES (2, repeat(_utf8'a', 128), 2);
+Comparing tables master:test.t4 and slave:test.t4
+[expecting slave to stop]
+INSERT INTO t5 VALUES (1, "", 1);
+INSERT INTO t5 VALUES (2, repeat(_utf8'a', 255), 2);
+Last_SQL_Error
+Table definition on master and slave does not match: Column 1 size mismatch - master has size 765, test.t5 on slave has size 49. Master's column size should be <= the slave's column size.
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+[expecting slave to stop]
+INSERT INTO t6 VALUES (1, "", 1);
+INSERT INTO t6 VALUES (2, repeat(_utf8'a', 255), 2);
+Last_SQL_Error
+Table definition on master and slave does not match: Column 1 size mismatch - master has size 765, test.t6 on slave has size 385. Master's column size should be <= the slave's column size.
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+[expecting slave to replicate correctly]
+INSERT INTO t7 VALUES (1, "", 1);
+INSERT INTO t7 VALUES (2, repeat(_utf8'a', 255), 2);
+Comparing tables master:test.t7 and slave:test.t7
+drop table t1, t2, t3, t4, t5, t6, t7;
diff --git a/mysql-test/suite/rpl/r/rpl_row_basic_3innodb.result b/mysql-test/suite/rpl/r/rpl_row_basic_3innodb.result
index f7b9e5d627a..d38d09fec00 100644
--- a/mysql-test/suite/rpl/r/rpl_row_basic_3innodb.result
+++ b/mysql-test/suite/rpl/r/rpl_row_basic_3innodb.result
@@ -437,7 +437,70 @@ SELECT COUNT(*) FROM t1 ORDER BY c1,c2;
 COUNT(*)	0
 set @@global.slave_exec_mode= default;
 Last_SQL_Error
-0
+
 SELECT COUNT(*) FROM t1 ORDER BY c1,c2;
 COUNT(*)	0
 DROP TABLE IF EXISTS t1,t2,t3,t4,t5,t6,t7,t8;
+CREATE TABLE t1 (i INT NOT NULL,
+c CHAR(16) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+CREATE TABLE t2 (i INT NOT NULL,
+c CHAR(16) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t2 MODIFY c CHAR(128) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t3 (i INT NOT NULL,
+c CHAR(128) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t3 MODIFY c CHAR(16) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t4 (i INT NOT NULL,
+c CHAR(128) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+CREATE TABLE t5 (i INT NOT NULL,
+c CHAR(255) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t5 MODIFY c CHAR(16) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t6 (i INT NOT NULL,
+c CHAR(255) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+ALTER TABLE t6 MODIFY c CHAR(128) CHARACTER SET utf8 NOT NULL;
+CREATE TABLE t7 (i INT NOT NULL,
+c CHAR(255) CHARACTER SET utf8 NOT NULL,
+j INT NOT NULL);
+[expecting slave to replicate correctly]
+INSERT INTO t1 VALUES (1, "", 1);
+INSERT INTO t1 VALUES (2, repeat(_utf8'a', 16), 2);
+Comparing tables master:test.t1 and slave:test.t1
+[expecting slave to replicate correctly]
+INSERT INTO t2 VALUES (1, "", 1);
+INSERT INTO t2 VALUES (2, repeat(_utf8'a', 16), 2);
+Comparing tables master:test.t2 and slave:test.t2
+[expecting slave to stop]
+INSERT INTO t3 VALUES (1, "", 1);
+INSERT INTO t3 VALUES (2, repeat(_utf8'a', 128), 2);
+Last_SQL_Error
+Table definition on master and slave does not match: Column 1 size mismatch - master has size 384, test.t3 on slave has size 49. Master's column size should be <= the slave's column size.
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+[expecting slave to replicate correctly]
+INSERT INTO t4 VALUES (1, "", 1);
+INSERT INTO t4 VALUES (2, repeat(_utf8'a', 128), 2);
+Comparing tables master:test.t4 and slave:test.t4
+[expecting slave to stop]
+INSERT INTO t5 VALUES (1, "", 1);
+INSERT INTO t5 VALUES (2, repeat(_utf8'a', 255), 2);
+Last_SQL_Error
+Table definition on master and slave does not match: Column 1 size mismatch - master has size 765, test.t5 on slave has size 49. Master's column size should be <= the slave's column size.
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+[expecting slave to stop]
+INSERT INTO t6 VALUES (1, "", 1);
+INSERT INTO t6 VALUES (2, repeat(_utf8'a', 255), 2);
+Last_SQL_Error
+Table definition on master and slave does not match: Column 1 size mismatch - master has size 765, test.t6 on slave has size 385. Master's column size should be <= the slave's column size.
+SET GLOBAL SQL_SLAVE_SKIP_COUNTER=8;
+START SLAVE;
+[expecting slave to replicate correctly]
+INSERT INTO t7 VALUES (1, "", 1);
+INSERT INTO t7 VALUES (2, repeat(_utf8'a', 255), 2);
+Comparing tables master:test.t7 and slave:test.t7
+drop table t1, t2, t3, t4, t5, t6, t7;
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