MDEV-20822 INET6 crashes in combination with RBR extended metadata

The code erroneously assumed that only Field_str descendants can
store character set information. After adding Field_inet6, it's
not true anymore.

Also, after adding Field_inet6, storing field->charset() become not correct either:
- Field_inet6::charset() return &my_charset_latin1,
  because clients see INET6 as VARCHAR(39).
- Field_inet6::binlog_type_info().m_cs returns &my_charset_bin
  because storage engines see INET6 as BINARY(16).

We need to store &my_charset_bin to the binlog metadata,
so the slave sees INET6 as BINARY(16), like storage engines do,
to make the slave treat the replicated data as binary IPv6 address
representation (rather than text representation).

The correct character set that needs to be stored to the metadata
is already populated to binlog_type_info_array[i].m_cs. So the
fixed code version uses this value rather than field->charset().

3 files changed, 136 insertions, 4 deletions

diff --git a/plugin/type_inet/mysql-test/type_inet/binlog_table_map_optional_metadata_type_inet6.result b/plugin/type_inet/mysql-test/type_inet/binlog_table_map_optional_metadata_type_inet6.result
new file mode 100644
index 00000000000..7911407b8f7
--- /dev/null
+++ b/plugin/type_inet/mysql-test/type_inet/binlog_table_map_optional_metadata_type_inet6.result
@@ -0,0 +1,60 @@
+#
+# Start of 10.5 tests
+#
+#
+# MDEV-20822 INET6 crashes in combination with RBR extended metadata
+#
+# Using DEFAULT_CHARSET format
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;
+CREATE TABLE t1 (a INET6);
+INSERT INTO t1 VALUES('::');
+# Columns(BINARY(16))
+DROP TABLE t1;
+RESET MASTER;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = MINIMAL;
+CREATE TABLE t1 (a INET6);
+INSERT INTO t1 VALUES('::');
+# Columns(BINARY(16))
+DROP TABLE t1;
+RESET MASTER;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+CREATE TABLE t1 (a INET6);
+INSERT INTO t1 VALUES('::');
+# Columns(`a` BINARY(16))
+DROP TABLE t1;
+RESET MASTER;
+# Using COLUMN_CHARSET format
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;
+CREATE TABLE t1 (a INET6, b CHAR(16) CHARACTER SET latin1, c CHAR(16) CHARACTER SET utf8);
+INSERT INTO t1 VALUES('::','','');
+# Columns(BINARY(16),
+#         BINARY(16),
+#         BINARY(48))
+DROP TABLE t1;
+RESET MASTER;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = MINIMAL;
+CREATE TABLE t1 (a INET6, b CHAR(16) CHARACTER SET latin1, c CHAR(16) CHARACTER SET utf8);
+INSERT INTO t1 VALUES('::','','');
+# Columns(BINARY(16),
+#         CHAR(16) CHARSET latin1 COLLATE latin1_swedish_ci,
+#         CHAR(16) CHARSET utf8 COLLATE utf8_general_ci)
+DROP TABLE t1;
+RESET MASTER;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+CREATE TABLE t1 (a INET6, b CHAR(16) CHARACTER SET latin1, c CHAR(16) CHARACTER SET utf8);
+INSERT INTO t1 VALUES('::','','');
+# Columns(`a` BINARY(16),
+#         `b` CHAR(16) CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c` CHAR(16) CHARSET utf8 COLLATE utf8_general_ci)
+DROP TABLE t1;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = DEFAULT;
+#
+# End of 10.5 tests
+#
diff --git a/plugin/type_inet/mysql-test/type_inet/binlog_table_map_optional_metadata_type_inet6.test b/plugin/type_inet/mysql-test/type_inet/binlog_table_map_optional_metadata_type_inet6.test
new file mode 100644
index 00000000000..63672f065a8
--- /dev/null
+++ b/plugin/type_inet/mysql-test/type_inet/binlog_table_map_optional_metadata_type_inet6.test
@@ -0,0 +1,72 @@
+--source include/have_debug.inc
+--source include/have_binlog_format_row.inc
+
+--let $MYSQLD_DATADIR= `select @@datadir`
+--let $binlog_file= $MYSQLD_DATADIR/master-bin.000001
+
+
+--echo #
+--echo # Start of 10.5 tests
+--echo #
+
+--echo #
+--echo # MDEV-20822 INET6 crashes in combination with RBR extended metadata
+--echo #
+
+--echo # Using DEFAULT_CHARSET format
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;
+CREATE TABLE t1 (a INET6);
+INSERT INTO t1 VALUES('::');
+--source suite/binlog/include/print_optional_metadata.inc
+DROP TABLE t1;
+RESET MASTER;
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = MINIMAL;
+CREATE TABLE t1 (a INET6);
+INSERT INTO t1 VALUES('::');
+--source suite/binlog/include/print_optional_metadata.inc
+DROP TABLE t1;
+RESET MASTER;
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+CREATE TABLE t1 (a INET6);
+INSERT INTO t1 VALUES('::');
+--source suite/binlog/include/print_optional_metadata.inc
+DROP TABLE t1;
+RESET MASTER;
+
+--echo # Using COLUMN_CHARSET format
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;
+CREATE TABLE t1 (a INET6, b CHAR(16) CHARACTER SET latin1, c CHAR(16) CHARACTER SET utf8);
+INSERT INTO t1 VALUES('::','','');
+--source suite/binlog/include/print_optional_metadata.inc
+DROP TABLE t1;
+RESET MASTER;
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = MINIMAL;
+CREATE TABLE t1 (a INET6, b CHAR(16) CHARACTER SET latin1, c CHAR(16) CHARACTER SET utf8);
+INSERT INTO t1 VALUES('::','','');
+--source suite/binlog/include/print_optional_metadata.inc
+DROP TABLE t1;
+RESET MASTER;
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+CREATE TABLE t1 (a INET6, b CHAR(16) CHARACTER SET latin1, c CHAR(16) CHARACTER SET utf8);
+INSERT INTO t1 VALUES('::','','');
+--source suite/binlog/include/print_optional_metadata.inc
+DROP TABLE t1;
+RESET MASTER;
+
+SET GLOBAL binlog_row_metadata = DEFAULT;
+
+--echo #
+--echo # End of 10.5 tests
+--echo #
diff --git a/sql/log_event_server.cc b/sql/log_event_server.cc
index 302610e2273..04cf70984a2 100644
--- a/sql/log_event_server.cc
+++ b/sql/log_event_server.cc
@@ -6562,12 +6562,12 @@ bool Table_map_log_event::init_charset_field(
     {
       if (include_type(binlog_type_info_array, m_table->field[i]))
       {
-        Field_str *field= dynamic_cast<Field_str *>(m_table->field[i]);
-
-        if (field->charset()->number != default_collation)
+        CHARSET_INFO *cs= binlog_type_info_array[i].m_cs;
+        DBUG_ASSERT(cs);
+        if (cs->number != default_collation)
         {
           store_compressed_length(buf, char_column_index);
-          store_compressed_length(buf, field->charset()->number);
+          store_compressed_length(buf, cs->number);
         }
         char_column_index++;
       }