Fix for BUG#51716 and BUG#51787.

In BUG#51787 we were using the wrong charset to print out the
data. We were using the field charset for the string that would
hold the information. This caused the assertion, because the
string length was not aligned with UTF32 bytes requirements for
storage.

We fix this by using &my_charset_latin1 in the string object
instead of the field->charset(). As a side-effect, we needed to
extend the show_sql_type interface so that it took the field
charset is now passed as a parameter, so that one is able to
calculate the correct field size.

In BUG#51716 we had issues with Field_string::pack and
Field_string::unpack. When packing, the length was incorrectly
calculated. When unpacking, the padding the string would be
padded with the wrong bytes (a few bytes less than it should).

We fix this by resorting to charset abstractions (functions) that
calculate the correct length when packing and pad correctly the
string when unpacking.

mysql-test/suite/rpl/r/rpl_row_charset.result

@@ -0,0 +1,42 @@
+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;
+SET SQL_LOG_BIN=0;
+CREATE TABLE t1 (c1 char(255) DEFAULT NULL, KEY c1 (c1)) DEFAULT CHARSET=utf32;
+Warnings:
+Warning	1071	Specified key was too long; max key length is 1000 bytes
+SET SQL_LOG_BIN=1;
+SET @saved_slave_type_conversions= @@global.slave_type_conversions;
+include/stop_slave.inc
+SET GLOBAL SLAVE_TYPE_CONVERSIONS='ALL_NON_LOSSY';
+include/start_slave.inc
+SET SQL_LOG_BIN=0;
+CREATE TABLE t1 ( c1 varchar(255) DEFAULT NULL, KEY c1 (c1)) DEFAULT CHARSET=utf32;
+Warnings:
+Warning	1071	Specified key was too long; max key length is 1000 bytes
+SET SQL_LOG_BIN=1;
+INSERT INTO t1(c1) VALUES ('insert into t1');
+DROP TABLE t1;
+SET GLOBAL SLAVE_TYPE_CONVERSIONS= @saved_slave_type_conversions;
+include/stop_slave.inc
+include/start_slave.inc
+CREATE TABLE t1(c1 CHAR(10) CHARACTER SET utf16 DEFAULT 'ola');
+INSERT INTO t1 VALUES ('abc');
+INSERT INTO t1 VALUES ();
+#### ON MASTER
+SELECT c1, hex(c1) from t1; ;
+c1	abc
+hex(c1)	006100620063
+c1	ola
+hex(c1)	006F006C0061
+#### ON SLAVE
+SELECT c1, hex(c1) FROM t1; ;
+c1	abc
+hex(c1)	006100620063
+c1	ola
+hex(c1)	006F006C0061
+Comparing tables master:test.t1 and slave:test.t1
+DROP TABLE t1;

mysql-test/suite/rpl/t/rpl_row_charset.test

@@ -0,0 +1,68 @@
+-- source include/master-slave.inc
+-- source include/have_binlog_format_row.inc
+
+#
+# BUG#51787 Assertion `(n % 4) == 0' on slave upon INSERT into a table with UTF32
+#
+
+SET SQL_LOG_BIN=0;
+CREATE TABLE t1 (c1 char(255) DEFAULT NULL, KEY c1 (c1)) DEFAULT CHARSET=utf32;
+SET SQL_LOG_BIN=1;
+
+-- connection slave
+
+-- let $reset_slave_type_conversions= 0
+
+SET @saved_slave_type_conversions= @@global.slave_type_conversions;
+
+#
+#  Force test to cover conversion execution path in the
+#  slave, which also makes use of sql_type method, thence
+#  can ultimately trigger the assertion.
+#
+-- source include/stop_slave.inc
+SET GLOBAL SLAVE_TYPE_CONVERSIONS='ALL_NON_LOSSY';
+-- source include/start_slave.inc
+
+SET SQL_LOG_BIN=0;
+CREATE TABLE t1 ( c1 varchar(255) DEFAULT NULL, KEY c1 (c1)) DEFAULT CHARSET=utf32;
+SET SQL_LOG_BIN=1;
+
+-- connection master
+
+INSERT INTO t1(c1) VALUES ('insert into t1');
+DROP TABLE t1;
+
+--sync_slave_with_master
+
+# assertion: the slave woul hit an/several assertions:
+#            before and during slave conversion procedure
+#            Now that is fixed, it wont.
+
+SET GLOBAL SLAVE_TYPE_CONVERSIONS= @saved_slave_type_conversions;
+-- source include/stop_slave.inc
+-- source include/start_slave.inc
+-- connection master
+
+#
+# BUG#51716: Char column with utf16 character set gives wrong padding on slave
+#
+
+CREATE TABLE t1(c1 CHAR(10) CHARACTER SET utf16 DEFAULT 'ola');
+INSERT INTO t1 VALUES ('abc');  # explicit value is inserted and encoded correctly
+INSERT INTO t1 VALUES ();       # default value is inserted and encoded correctly
+
+-- echo #### ON MASTER
+--query_vertical SELECT c1, hex(c1) from t1; 
+
+-- sync_slave_with_master
+
+-- echo #### ON SLAVE
+--query_vertical SELECT c1, hex(c1) FROM t1; 
+
+# assertion: tables don't differ
+-- let $diff_table_1=master:test.t1
+-- let $diff_table_2=slave:test.t1
+-- source include/diff_tables.inc
+
+DROP TABLE t1;

sql/field.cc

@@ -6613,8 +6613,7 @@ uchar *Field_string::pack(uchar *to, const uchar *from,
     local_char_length= my_charpos(field_charset, from, from+length,
                                   local_char_length);
   set_if_smaller(length, local_char_length);
-  while (length && from[length-1] == field_charset->pad_char)
+  length= field_charset->cset->lengthsp(field_charset, (const char*) from, length);
-    length--;
 
   // Length always stored little-endian
   *to++= (uchar) length;
@@ -6680,7 +6679,7 @@ Field_string::unpack(uchar *to,
 
   memcpy(to, from, length);
   // Pad the string with the pad character of the fields charset
-  bfill(to + length, field_length - length, field_charset->pad_char);
+  field_charset->cset->fill(field_charset, (char*) to + length, field_length - length, field_charset->pad_char);
   return from+length;
 }
 

sql/rpl_utility.cc

@@ -340,7 +340,7 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
 
 /**
  */
-void show_sql_type(enum_field_types type, uint16 metadata, String *str)
+void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_INFO *field_cs)
 {
   DBUG_ENTER("show_sql_type");
   DBUG_PRINT("enter", ("type: %d, metadata: 0x%x", type, metadata));
@@ -489,7 +489,7 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str)
       uint bytes= (((metadata >> 4) & 0x300) ^ 0x300) + (metadata & 0x00ff);
       uint32 length=
         cs->cset->snprintf(cs, (char*) str->ptr(), str->alloced_length(),
-                           "char(%d)", bytes / cs->mbmaxlen);
+                           "char(%d)", bytes / field_cs->mbmaxlen);
       str->length(length);
     }
     break;
@@ -579,7 +579,7 @@ can_convert_field_to(Field *field,
   DBUG_ENTER("can_convert_field_to");
 #ifndef DBUG_OFF
   char field_type_buf[MAX_FIELD_WIDTH];
-  String field_type(field_type_buf, sizeof(field_type_buf), field->charset());
+  String field_type(field_type_buf, sizeof(field_type_buf), &my_charset_latin1);
   field->sql_type(field_type);
   DBUG_PRINT("enter", ("field_type: %s, target_type: %d, source_type: %d, source_metadata: 0x%x",
                        field_type.c_ptr_safe(), field->real_type(), source_type, metadata));
@@ -822,9 +822,9 @@ table_def::compatible_with(THD *thd, Relay_log_info *rli,
       const char *tbl_name= table->s->table_name.str;
       char source_buf[MAX_FIELD_WIDTH];
       char target_buf[MAX_FIELD_WIDTH];
-      String source_type(source_buf, sizeof(source_buf), field->charset());
+      String source_type(source_buf, sizeof(source_buf), &my_charset_latin1);
-      String target_type(target_buf, sizeof(target_buf), field->charset());
+      String target_type(target_buf, sizeof(target_buf), &my_charset_latin1);
-      show_sql_type(type(col), field_metadata(col), &source_type);
+      show_sql_type(type(col), field_metadata(col), &source_type, field->charset());
       field->sql_type(target_type);
       rli->report(ERROR_LEVEL, ER_SLAVE_CONVERSION_FAILED,
                   ER(ER_SLAVE_CONVERSION_FAILED),
@@ -842,8 +842,8 @@ table_def::compatible_with(THD *thd, Relay_log_info *rli,
       {
         char source_buf[MAX_FIELD_WIDTH];
         char target_buf[MAX_FIELD_WIDTH];
-        String source_type(source_buf, sizeof(source_buf), table->field[col]->charset());
+        String source_type(source_buf, sizeof(source_buf), &my_charset_latin1);
-        String target_type(target_buf, sizeof(target_buf), table->field[col]->charset());
+        String target_type(target_buf, sizeof(target_buf), &my_charset_latin1);
         tmp_table->field[col]->sql_type(source_type);
         table->field[col]->sql_type(target_type);
         DBUG_PRINT("debug", ("Field %s - conversion required."