MDEV-20477 Merge binlog extended metadata support from the upstream

Cherry-pick the commits the mysql and some changes.
WL#4618 RBR: extended table metadata in the binary log

This patch extends Table Map Event. It appends some new fields for
more metadata. The new metadata includes:
- Signedness of Numberic Columns
- Character Set of Character Columns and Binary Columns
- Column Name
- String Value of SET Columns
- String Value of ENUM Columns
- Primary Key
- Character Set of SET Columns and ENUM Columns
- Geometry Type

Some of them are optional, the patch introduces a GLOBAL system
variable to control it. It is binlog_row_metadata.
- Scope:   GLOBAL
- Dynamic: Yes
- Type:    ENUM
- Values:  {NO_LOG, MINIMAL, FULL}
- Default: NO_LOG
  Only Signedness, character set and geometry type are logged if it is MINIMAL.
  Otherwise all of them are logged.

Also add a binlog_type_info() to field, So that we can have extract
relevant binlog info from field.

client/client_priv.h

@@ -102,6 +102,7 @@ enum options_client
   OPT_PRINT_ROW_COUNT, OPT_PRINT_ROW_EVENT_POSITIONS,
   OPT_SHUTDOWN_WAIT_FOR_SLAVES,
   OPT_COPY_S3_TABLES,
+  OPT_PRINT_TABLE_METADATA,
   OPT_MAX_CLIENT_OPTION /* should be always the last */
 };
 

client/mysqlbinlog.cc

@@ -144,6 +144,7 @@ static const char* dirname_for_local_load= 0;
 static bool opt_skip_annotate_row_events= 0;
 
 static my_bool opt_flashback;
+static bool opt_print_table_metadata;
 #ifdef WHEN_FLASHBACK_REVIEW_READY
 static my_bool opt_flashback_review;
 static char *flashback_review_dbname, *flashback_review_tablename;
@@ -1095,6 +1096,7 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
       print_event_info->hexdump_from= pos;
 
     print_event_info->base64_output_mode= opt_base64_output_mode;
+    print_event_info->print_table_metadata= opt_print_table_metadata;
 
     DBUG_PRINT("debug", ("event_type: %s", ev->get_type_str()));
 
@@ -1788,6 +1790,10 @@ Example: rewrite-db='from->to'.",
    (uchar**) &opt_skip_annotate_row_events,
    (uchar**) &opt_skip_annotate_row_events,
    0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"print-table-metadata", OPT_PRINT_TABLE_METADATA,
+   "Print metadata stored in Table_map_log_event",
+   &opt_print_table_metadata, &opt_print_table_metadata, 0,
+   GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
   {0, 0, 0, 0, 0, 0, GET_NO_ARG, NO_ARG, 0, 0, 0, 0, 0, 0}
 };
 

mysql-test/include/search_pattern_in_file.inc

@@ -9,6 +9,9 @@
 #
 #    The environment variables SEARCH_FILE and SEARCH_PATTERN must be set
 #    before sourcing this routine.
+#    SEARCH_TYPE can also be set to either NULL(default) or _gm_
+#    NULL is equivalent of using m/SEARCH_PATTERN/gs
+#    _gm_ is equivalent of using m/SEARCH_RANGE/gm
 #
 #    Optionally, SEARCH_RANGE can be set to the max number of bytes of the file
 #    to search. If negative, it will search that many bytes at the end of the
@@ -77,7 +80,15 @@ perl;
         close(FILE);
         $content.= $file_content;
     }
-    my @matches=($content =~ m/$search_pattern/gs);
+    my @matches;
+    if (not defined($ENV{SEARCH_TYPE}))
+    {
+      @matches=($content =~ /$search_pattern/gs);
+    }
+    elsif($ENV{SEARCH_TYPE} == "_gm_")
+    {
+      @matches=($content =~ /$search_pattern/gm);
+    }
     my $res=@matches ? "FOUND " . scalar(@matches) : "NOT FOUND";
     $ENV{SEARCH_FILE} =~ s{^.*?([^/\\]+)$}{$1};
 

mysql-test/main/mysqld--help.result

@@ -114,6 +114,12 @@ The following specify which files/extra groups are read (specified before remain
  the table) is logged in the before image, and only
  changed columns are logged in the after image. (Default:
  FULL).
+ --binlog-row-metadata=name 
+ Controls whether metadata is logged using FULL , MINIMAL
+ format and NO_LOG.FULL causes all metadata to be logged;
+ MINIMAL means that only metadata actually required by
+ slave is logged; NO_LOG NO metadata will be
+ logged.Default: NO_LOG.
  --binlog-stmt-cache-size=# 
  The size of the statement cache for updates to
  non-transactional engines for the binary log. If you
@@ -1433,6 +1439,7 @@ binlog-format MIXED
 binlog-optimize-thread-scheduling TRUE
 binlog-row-event-max-size 8192
 binlog-row-image FULL
+binlog-row-metadata NO_LOG
 binlog-stmt-cache-size 32768
 bulk-insert-buffer-size 8388608
 character-set-client-handshake TRUE

mysql-test/suite/binlog/include/print_optional_metadata.inc

+# Auxaliary file for printing optional metadata in table_map_log_event
+# Usage :
+# --let $binlog_file=
+# [--let $stop_position]
+# [--let $print_primary_key]
+# --source extra/binlog_tests/print_optional_metadata.inc
+
+--let $output_file= $MYSQLTEST_VARDIR/tmp/mysqlbinlog.output
+
+--let $_stop_position_opt=
+if ($stop_position)
+{
+  --let $_stop_position_opt=--stop-position=$stop_position
+}
+
+--exec $MYSQL_BINLOG -F --print-table-metadata $_stop_position_opt $binlog_file > $output_file
+
+
+--let SEARCH_PATTERN= # (?:Columns\(| {8}).*
+--let SEARCH_FILE= $output_file
+--let SEARCH_OUTPUT=matches
+--let SEARCH_TYPE="_gm_"
+--source include/search_pattern_in_file.inc
+
+if ($print_primary_key)
+{
+  --let SEARCH_PATTERN= # Primary Key
+  --source include/search_pattern_in_file.inc
+}
+--remove_file $output_file
+--let $stop_position=
+--let $_stop_position_opt=
+
+

mysql-test/suite/binlog/r/binlog_table_map_optional_metadata_binary.result

+#
+# Verify that SET string values and character sets can be printed correctly
+#
+SET NAMES utf8;
+CREATE TABLE t1(
+c_set_1 SET("set1_v1_å", "set1_v2_ä", "set1_v3_ö"),
+c_set_2 SET("set2_v1_å", "set2_v2_ä", "set2_v3_ö") CHARACTER SET binary);
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+# Columns(SET,
+#         SET)
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+# Columns(`c_set_1` SET('set1_v1_','set1_v2_','set1_v3_') CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c_set_2` SET('set2_v1_å','set2_v2_ä','set2_v3_ö') CHARSET binary COLLATE binary)
+INSERT INTO t1 VALUES("set1_v3_ö", "set2_v3_ö");
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v1_å");
+SELECT c_set_1, HEX(c_set_1) FROM t1;
+c_set_1	HEX(c_set_1)
+set1_v1_å	736574315F76315FE5
+set1_v1_å	736574315F76315FE5
+set1_v3_ö	736574315F76335FF6
+set1_v1_å	736574315F76315FE5
+SELECT c_set_2, HEX(c_set_2) FROM t1;
+c_set_2	HEX(c_set_2)
+set2_v2_ä	736574325F76325FC3A4
+set2_v2_ä	736574325F76325FC3A4
+set2_v3_ö	736574325F76335FC3B6
+set2_v1_å	736574325F76315FC3A5
+DROP TABLE t1;
+RESET MASTER;
+#
+# Verify that ENUM string values and character sets can be printed correctly
+#
+CREATE TABLE t1(
+c_enum_1 ENUM("enum1_v1_å", "enum1_v2_ä", "enum1_v3_ö"),
+c_enum_2 ENUM("enum2_v1_å", "enum2_v2_ä", "enum2_v3_ö") CHARACTER SET binary);
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+# Columns(ENUM,
+#         ENUM)
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+# Columns(`c_enum_1` ENUM('enum1_v1_','enum1_v2_','enum1_v3_') CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c_enum_2` ENUM('enum2_v1_å','enum2_v2_ä','enum2_v3_ö') CHARSET binary COLLATE binary)
+INSERT INTO t1 VALUES("enum1_v3_ö", "enum2_v3_ö");
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v1_å");
+SELECT c_enum_1, HEX(c_enum_1) FROM t1;
+c_enum_1	HEX(c_enum_1)
+enum1_v1_å	656E756D315F76315FE5
+enum1_v1_å	656E756D315F76315FE5
+enum1_v3_ö	656E756D315F76335FF6
+enum1_v1_å	656E756D315F76315FE5
+SELECT c_enum_2, HEX(c_enum_2) FROM t1;
+c_enum_2	HEX(c_enum_2)
+enum2_v2_ä	656E756D325F76325FC3A4
+enum2_v2_ä	656E756D325F76325FC3A4
+enum2_v3_ö	656E756D325F76335FC3B6
+enum2_v1_å	656E756D325F76315FC3A5
+DROP TABLE t1;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;

mysql-test/suite/binlog/r/binlog_table_map_optional_metadata_ucs2.result

+#
+# Verify that SET string values and character sets can be printed correctly
+#
+SET NAMES utf8;
+CREATE TABLE t1(
+c_set_1 SET("set1_v1_å", "set1_v2_ä", "set1_v3_ö"),
+c_set_2 SET("set2_v1_å", "set2_v2_ä", "set2_v3_ö") CHARACTER SET ucs2);
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+# Columns(SET,
+#         SET)
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+# Columns(`c_set_1` SET('set1_v1_','set1_v2_','set1_v3_') CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c_set_2` SET('\0s\0e\0t\02\0_\0v\01\0_\0','\0s\0e\0t\02\0_\0v\02\0_\0','\0s\0e\0t\02\0_\0v\03\0_\0') CHARSET ucs2 COLLATE ucs2_general_ci)
+INSERT INTO t1 VALUES("set1_v3_ö", "set2_v3_ö");
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v1_å");
+SELECT c_set_1, HEX(c_set_1) FROM t1;
+c_set_1	HEX(c_set_1)
+set1_v1_å	736574315F76315FE5
+set1_v1_å	736574315F76315FE5
+set1_v3_ö	736574315F76335FF6
+set1_v1_å	736574315F76315FE5
+SELECT c_set_2, HEX(c_set_2) FROM t1;
+c_set_2	HEX(c_set_2)
+set2_v2_ä	0073006500740032005F00760032005F00E4
+set2_v2_ä	0073006500740032005F00760032005F00E4
+set2_v3_ö	0073006500740032005F00760033005F00F6
+set2_v1_å	0073006500740032005F00760031005F00E5
+DROP TABLE t1;
+RESET MASTER;
+#
+# Verify that ENUM string values and character sets can be printed correctly
+#
+CREATE TABLE t1(
+c_enum_1 ENUM("enum1_v1_å", "enum1_v2_ä", "enum1_v3_ö"),
+c_enum_2 ENUM("enum2_v1_å", "enum2_v2_ä", "enum2_v3_ö") CHARACTER SET ucs2);
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+# Columns(ENUM,
+#         ENUM)
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+# Columns(`c_enum_1` ENUM('enum1_v1_','enum1_v2_','enum1_v3_') CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c_enum_2` ENUM('\0e\0n\0u\0m\02\0_\0v\01\0_\0','\0e\0n\0u\0m\02\0_\0v\02\0_\0','\0e\0n\0u\0m\02\0_\0v\03\0_\0') CHARSET ucs2 COLLATE ucs2_general_ci)
+INSERT INTO t1 VALUES("enum1_v3_ö", "enum2_v3_ö");
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v1_å");
+SELECT c_enum_1, HEX(c_enum_1) FROM t1;
+c_enum_1	HEX(c_enum_1)
+enum1_v1_å	656E756D315F76315FE5
+enum1_v1_å	656E756D315F76315FE5
+enum1_v3_ö	656E756D315F76335FF6
+enum1_v1_å	656E756D315F76315FE5
+SELECT c_enum_2, HEX(c_enum_2) FROM t1;
+c_enum_2	HEX(c_enum_2)
+enum2_v2_ä	0065006E0075006D0032005F00760032005F00E4
+enum2_v2_ä	0065006E0075006D0032005F00760032005F00E4
+enum2_v3_ö	0065006E0075006D0032005F00760033005F00F6
+enum2_v1_å	0065006E0075006D0032005F00760031005F00E5
+DROP TABLE t1;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;

mysql-test/suite/binlog/r/binlog_table_map_optional_metadata_utf32.result

+#
+# Verify that SET string values and character sets can be printed correctly
+#
+SET NAMES utf8;
+CREATE TABLE t1(
+c_set_1 SET("set1_v1_å", "set1_v2_ä", "set1_v3_ö"),
+c_set_2 SET("set2_v1_å", "set2_v2_ä", "set2_v3_ö") CHARACTER SET utf32);
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+# Columns(SET,
+#         SET)
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+# Columns(`c_set_1` SET('set1_v1_','set1_v2_','set1_v3_') CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c_set_2` SET('\0\0\0s\0\0\0e\0\0\0t\0\0\02\0\0\0_\0\0\0v\0\0\01\0\0\0_\0\0\0','\0\0\0s\0\0\0e\0\0\0t\0\0\02\0\0\0_\0\0\0v\0\0\02\0\0\0_\0\0\0','\0\0\0s\0\0\0e\0\0\0t\0\0\02\0\0\0_\0\0\0v\0\0\03\0\0\0_\0\0\0') CHARSET utf32 COLLATE utf32_general_ci)
+INSERT INTO t1 VALUES("set1_v3_ö", "set2_v3_ö");
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v1_å");
+SELECT c_set_1, HEX(c_set_1) FROM t1;
+c_set_1	HEX(c_set_1)
+set1_v1_å	736574315F76315FE5
+set1_v1_å	736574315F76315FE5
+set1_v3_ö	736574315F76335FF6
+set1_v1_å	736574315F76315FE5
+SELECT c_set_2, HEX(c_set_2) FROM t1;
+c_set_2	HEX(c_set_2)
+set2_v2_ä	000000730000006500000074000000320000005F00000076000000320000005F000000E4
+set2_v2_ä	000000730000006500000074000000320000005F00000076000000320000005F000000E4
+set2_v3_ö	000000730000006500000074000000320000005F00000076000000330000005F000000F6
+set2_v1_å	000000730000006500000074000000320000005F00000076000000310000005F000000E5
+DROP TABLE t1;
+RESET MASTER;
+#
+# Verify that ENUM string values and character sets can be printed correctly
+#
+CREATE TABLE t1(
+c_enum_1 ENUM("enum1_v1_å", "enum1_v2_ä", "enum1_v3_ö"),
+c_enum_2 ENUM("enum2_v1_å", "enum2_v2_ä", "enum2_v3_ö") CHARACTER SET utf32);
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+# Columns(ENUM,
+#         ENUM)
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+# Columns(`c_enum_1` ENUM('enum1_v1_','enum1_v2_','enum1_v3_') CHARSET latin1 COLLATE latin1_swedish_ci,
+#         `c_enum_2` ENUM('\0\0\0e\0\0\0n\0\0\0u\0\0\0m\0\0\02\0\0\0_\0\0\0v\0\0\01\0\0\0_\0\0\0','\0\0\0e\0\0\0n\0\0\0u\0\0\0m\0\0\02\0\0\0_\0\0\0v\0\0\02\0\0\0_\0\0\0','\0\0\0e\0\0\0n\0\0\0u\0\0\0m\0\0\02\0\0\0_\0\0\0v\0\0\03\0\0\0_\0\0\0') CHARSET utf32 COLLATE utf32_general_ci)
+INSERT INTO t1 VALUES("enum1_v3_ö", "enum2_v3_ö");
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v1_å");
+SELECT c_enum_1, HEX(c_enum_1) FROM t1;
+c_enum_1	HEX(c_enum_1)
+enum1_v1_å	656E756D315F76315FE5
+enum1_v1_å	656E756D315F76315FE5
+enum1_v3_ö	656E756D315F76335FF6
+enum1_v1_å	656E756D315F76315FE5
+SELECT c_enum_2, HEX(c_enum_2) FROM t1;
+c_enum_2	HEX(c_enum_2)
+enum2_v2_ä	000000650000006E000000750000006D000000320000005F00000076000000320000005F000000E4
+enum2_v2_ä	000000650000006E000000750000006D000000320000005F00000076000000320000005F000000E4
+enum2_v3_ö	000000650000006E000000750000006D000000320000005F00000076000000330000005F000000F6
+enum2_v1_å	000000650000006E000000750000006D000000320000005F00000076000000310000005F000000E5
+DROP TABLE t1;
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = NO_LOG;

mysql-test/suite/binlog/t/binlog_table_map_optional_metadata_binary.test

+################################################################################
+# WL#4618 RBR: extended table metadata in the binary log
+#
+# Below metadata is logged into Table_map_log_event
+# - signedness of numeric columns
+# - charsets of character columns
+# - column names
+# - set/enum character sets and string values
+# - primary key
+#
+# The first two are always logged. The others are controlled by system
+# variable --binlog-row-metadata
+#
+# The test will verify if the metadata can be logged and printed by mysqlbinlog
+# correctly.
+# mysqlbinlog --print-table-metadata will print the extra metadata
+################################################################################
+--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 # Verify that SET string values and character sets can be printed correctly
+--echo #
+
+SET NAMES utf8;
+CREATE TABLE t1(
+  c_set_1 SET("set1_v1_å", "set1_v2_ä", "set1_v3_ö"),
+  c_set_2 SET("set2_v1_å", "set2_v2_ä", "set2_v3_ö") CHARACTER SET binary);
+
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+--source include/print_optional_metadata.inc
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+--source include/print_optional_metadata.inc
+INSERT INTO t1 VALUES("set1_v3_ö", "set2_v3_ö");
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v1_å");
+SELECT c_set_1, HEX(c_set_1) FROM t1;
+SELECT c_set_2, HEX(c_set_2) FROM t1;
+
+DROP TABLE t1;
+RESET MASTER;
+
+--echo #
+--echo # Verify that ENUM string values and character sets can be printed correctly
+--echo #
+
+CREATE TABLE t1(
+  c_enum_1 ENUM("enum1_v1_å", "enum1_v2_ä", "enum1_v3_ö"),
+  c_enum_2 ENUM("enum2_v1_å", "enum2_v2_ä", "enum2_v3_ö") CHARACTER SET binary);
+
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+--source include/print_optional_metadata.inc
+
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+--source include/print_optional_metadata.inc
+INSERT INTO t1 VALUES("enum1_v3_ö", "enum2_v3_ö");
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v1_å");
+SELECT c_enum_1, HEX(c_enum_1) FROM t1;
+SELECT c_enum_2, HEX(c_enum_2) FROM t1;
+
+DROP TABLE t1;
+RESET MASTER;
+
+SET GLOBAL binlog_row_metadata = NO_LOG;

mysql-test/suite/binlog/t/binlog_table_map_optional_metadata_ucs2.test

+################################################################################
+# WL#4618 RBR: extended table metadata in the binary log
+#
+# Below metadata is logged into Table_map_log_event
+# - signedness of numeric columns
+# - charsets of character columns
+# - column names
+# - set/enum character sets and string values
+# - primary key
+#
+# The first two are always logged. The others are controlled by system
+# variable --binlog-row-metadata
+#
+# The test will verify if the metadata can be logged and printed by mysqlbinlog
+# correctly.
+# mysqlbinlog --print-table-metadata will print the extra metadata
+################################################################################
+--source include/have_debug.inc
+--source include/have_binlog_format_row.inc
+--source include/have_ucs2.inc
+--let $MYSQLD_DATADIR= `select @@datadir`
+--let $binlog_file= $MYSQLD_DATADIR/master-bin.000001
+
+--echo #
+--echo # Verify that SET string values and character sets can be printed correctly
+--echo #
+
+SET NAMES utf8;
+CREATE TABLE t1(
+  c_set_1 SET("set1_v1_å", "set1_v2_ä", "set1_v3_ö"),
+  c_set_2 SET("set2_v1_å", "set2_v2_ä", "set2_v3_ö") CHARACTER SET ucs2);
+
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+--source include/print_optional_metadata.inc
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+--source include/print_optional_metadata.inc
+INSERT INTO t1 VALUES("set1_v3_ö", "set2_v3_ö");
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v1_å");
+SELECT c_set_1, HEX(c_set_1) FROM t1;
+SELECT c_set_2, HEX(c_set_2) FROM t1;
+
+DROP TABLE t1;
+RESET MASTER;
+
+--echo #
+--echo # Verify that ENUM string values and character sets can be printed correctly
+--echo #
+
+CREATE TABLE t1(
+  c_enum_1 ENUM("enum1_v1_å", "enum1_v2_ä", "enum1_v3_ö"),
+  c_enum_2 ENUM("enum2_v1_å", "enum2_v2_ä", "enum2_v3_ö") CHARACTER SET ucs2);
+
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+--source include/print_optional_metadata.inc
+
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+--source include/print_optional_metadata.inc
+INSERT INTO t1 VALUES("enum1_v3_ö", "enum2_v3_ö");
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v1_å");
+SELECT c_enum_1, HEX(c_enum_1) FROM t1;
+SELECT c_enum_2, HEX(c_enum_2) FROM t1;
+
+DROP TABLE t1;
+RESET MASTER;
+
+SET GLOBAL binlog_row_metadata = NO_LOG;

mysql-test/suite/binlog/t/binlog_table_map_optional_metadata_utf32.test

+################################################################################
+# WL#4618 RBR: extended table metadata in the binary log
+#
+# Below metadata is logged into Table_map_log_event
+# - signedness of numeric columns
+# - charsets of character columns
+# - column names
+# - set/enum character sets and string values
+# - primary key
+#
+# The first two are always logged. The others are controlled by system
+# variable --binlog-row-metadata
+#
+# The test will verify if the metadata can be logged and printed by mysqlbinlog
+# correctly.
+# mysqlbinlog --print-table-metadata will print the extra metadata
+################################################################################
+--source include/have_debug.inc
+--source include/have_binlog_format_row.inc
+--source include/have_utf32.inc
+--let $MYSQLD_DATADIR= `select @@datadir`
+--let $binlog_file= $MYSQLD_DATADIR/master-bin.000001
+
+--echo #
+--echo # Verify that SET string values and character sets can be printed correctly
+--echo #
+
+SET NAMES utf8;
+CREATE TABLE t1(
+  c_set_1 SET("set1_v1_å", "set1_v2_ä", "set1_v3_ö"),
+  c_set_2 SET("set2_v1_å", "set2_v2_ä", "set2_v3_ö") CHARACTER SET utf32);
+
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+--source include/print_optional_metadata.inc
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v2_ä");
+--source include/print_optional_metadata.inc
+INSERT INTO t1 VALUES("set1_v3_ö", "set2_v3_ö");
+INSERT INTO t1 VALUES("set1_v1_å", "set2_v1_å");
+SELECT c_set_1, HEX(c_set_1) FROM t1;
+SELECT c_set_2, HEX(c_set_2) FROM t1;
+
+DROP TABLE t1;
+RESET MASTER;
+
+--echo #
+--echo # Verify that ENUM string values and character sets can be printed correctly
+--echo #
+
+CREATE TABLE t1(
+  c_enum_1 ENUM("enum1_v1_å", "enum1_v2_ä", "enum1_v3_ö"),
+  c_enum_2 ENUM("enum2_v1_å", "enum2_v2_ä", "enum2_v3_ö") CHARACTER SET utf32);
+
+SET GLOBAL binlog_row_metadata = MINIMAL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+--source include/print_optional_metadata.inc
+
+
+RESET MASTER;
+SET GLOBAL binlog_row_metadata = FULL;
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v2_ä");
+--source include/print_optional_metadata.inc
+INSERT INTO t1 VALUES("enum1_v3_ö", "enum2_v3_ö");
+INSERT INTO t1 VALUES("enum1_v1_å", "enum2_v1_å");
+SELECT c_enum_1, HEX(c_enum_1) FROM t1;
+SELECT c_enum_2, HEX(c_enum_2) FROM t1;
+
+DROP TABLE t1;
+RESET MASTER;
+
+SET GLOBAL binlog_row_metadata = NO_LOG;

mysql-test/suite/sys_vars/r/binlog_row_metadata_basic.result

+NO_LOG Expected
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+NO_LOG
+SELECT @@SESSION.binlog_row_metadata;
+ERROR HY000: Variable 'binlog_row_metadata' is a GLOBAL variable
+'#---------------------BS_STVARS_002_01----------------------#'
+SET @start_value= @@global.binlog_row_metadata;
+SELECT COUNT(@@GLOBAL.binlog_row_metadata);
+COUNT(@@GLOBAL.binlog_row_metadata)
+1
+1 Expected
+'#---------------------BS_STVARS_002_02----------------------#'
+SET @@GLOBAL.binlog_row_metadata=0;
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+NO_LOG
+NO_LOG Expected
+SET @@GLOBAL.binlog_row_metadata=1;
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+MINIMAL
+MINIMAL Expected
+SET @@GLOBAL.binlog_row_metadata=2;
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+FULL
+FULL Expected
+SET @@GLOBAL.binlog_row_metadata=NO_LOG;
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+NO_LOG
+NO_LOG Expected
+SET @@GLOBAL.binlog_row_metadata=MINIMAL;
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+MINIMAL
+MINIMAL Expected
+SET @@GLOBAL.binlog_row_metadata=FULL;
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+FULL
+FULL Expected
+SET @@GLOBAL.binlog_row_metadata='NO_LOG';
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+NO_LOG
+NO_LOG Expected
+SET @@GLOBAL.binlog_row_metadata='MINIMAL';
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+MINIMAL
+MINIMAL Expected
+SET @@GLOBAL.binlog_row_metadata='FULL';
+SELECT @@GLOBAL.binlog_row_metadata;
+@@GLOBAL.binlog_row_metadata
+FULL
+FULL Expected
+'#---------------------BS_STVARS_002_03----------------------#'
+SET @@GLOBAL.binlog_row_metadata='MINIMAl';
+SELECT *
+FROM information_schema.global_variables
+WHERE VARIABLE_NAME='binlog_row_metadata';
+VARIABLE_NAME	VARIABLE_VALUE
+BINLOG_ROW_METADATA	MINIMAL
+'#---------------------BS_STVARS_002_04----------------------#'
+SELECT *
+FROM information_schema.session_variables
+WHERE VARIABLE_NAME LIKE 'binlog_row_metadata';
+VARIABLE_NAME	VARIABLE_VALUE
+BINLOG_ROW_METADATA	MINIMAL
+'#---------------------BS_STVARS_002_05----------------------#'
+SELECT COUNT(@@binlog_row_metadata);
+COUNT(@@binlog_row_metadata)
+1
+1 Expected
+SELECT COUNT(@@GLOBAL.binlog_row_metadata);
+COUNT(@@GLOBAL.binlog_row_metadata)
+1
+1 Expected
+'#---------------------BS_STVARS_002_06----------------------#'
+SET GLOBAL binlog_row_metadata = full1;
+ERROR 42000: Variable 'binlog_row_metadata' can't be set to the value of 'full1'
+SET GLOBAL binlog_row_metadata = "full1";
+ERROR 42000: Variable 'binlog_row_metadata' can't be set to the value of 'full1'
+SET GLOBAL binlog_row_metadata = 3;
+ERROR 42000: Variable 'binlog_row_metadata' can't be set to the value of '3'
+SET GLOBAL binlog_row_metadata = -1;
+ERROR 42000: Variable 'binlog_row_metadata' can't be set to the value of '-1'
+SET @@global.binlog_row_metadata= @start_value;

mysql-test/suite/sys_vars/r/sysvars_server_notembedded.result

@@ -423,6 +423,20 @@ NUMERIC_BLOCK_SIZE	NULL
 ENUM_VALUE_LIST	MINIMAL,NOBLOB,FULL
 READ_ONLY	NO
 COMMAND_LINE_ARGUMENT	REQUIRED
+VARIABLE_NAME	BINLOG_ROW_METADATA
+SESSION_VALUE	NULL
+GLOBAL_VALUE	NO_LOG
+GLOBAL_VALUE_ORIGIN	COMPILE-TIME
+DEFAULT_VALUE	NO_LOG
+VARIABLE_SCOPE	GLOBAL
+VARIABLE_TYPE	ENUM
+VARIABLE_COMMENT	Controls whether metadata is logged using FULL , MINIMAL format and NO_LOG.FULL causes all metadata to be logged; MINIMAL means that only metadata actually required by slave is logged; NO_LOG NO metadata will be logged.Default: NO_LOG.
+NUMERIC_MIN_VALUE	NULL
+NUMERIC_MAX_VALUE	NULL
+NUMERIC_BLOCK_SIZE	NULL
+ENUM_VALUE_LIST	NO_LOG,MINIMAL,FULL
+READ_ONLY	NO
+COMMAND_LINE_ARGUMENT	REQUIRED
 VARIABLE_NAME	BINLOG_STMT_CACHE_SIZE
 VARIABLE_SCOPE	GLOBAL
 VARIABLE_TYPE	BIGINT UNSIGNED

mysql-test/suite/sys_vars/t/binlog_row_metadata_basic.test

+################## mysql-test\t\binlog_row_metadata_basic.test ################
+#                                                                             #
+# Variable Name: binlog_row_metadata                                          #
+# Scope: Global                                                               #
+# Data Type: enumeration                                                      #
+#                                                                             #
+# Creation Date: 2017-01-23                                                   #
+# Author : Libing Song                                                        #
+#                                                                             #
+#                                                                             #
+# Description:Test Cases of Dynamic System Variable binlog_row_metadata       #
+#             that checks the behavior of this variable in the following ways #
+#              * Value Check                                                  #
+#              * Scope Check                                                  #
+#                                                                             #
+# Reference:                                                                  #
+#   http://dev.mysql.com/doc/refman/8.X/en/server-system-variables.html       #
+#                                                                             #
+###############################################################################
+
+
+--echo NO_LOG Expected
+SELECT @@GLOBAL.binlog_row_metadata;
+--error ER_INCORRECT_GLOBAL_LOCAL_VAR
+SELECT @@SESSION.binlog_row_metadata;
+
+--echo '#---------------------BS_STVARS_002_01----------------------#'
+####################################################################
+#   Displaying default value                                       #
+####################################################################
+SET @start_value= @@global.binlog_row_metadata;
+
+SELECT COUNT(@@GLOBAL.binlog_row_metadata);
+--echo 1 Expected
+
+--echo '#---------------------BS_STVARS_002_02----------------------#'
+####################################################################
+#   Check if Value can set                                         #
+####################################################################
+SET @@GLOBAL.binlog_row_metadata=0;
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo NO_LOG Expected
+
+SET @@GLOBAL.binlog_row_metadata=1;
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo MINIMAL Expected
+
+SET @@GLOBAL.binlog_row_metadata=2;
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo FULL Expected
+
+SET @@GLOBAL.binlog_row_metadata=NO_LOG;
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo NO_LOG Expected
+
+SET @@GLOBAL.binlog_row_metadata=MINIMAL;
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo MINIMAL Expected
+
+SET @@GLOBAL.binlog_row_metadata=FULL;
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo FULL Expected
+
+SET @@GLOBAL.binlog_row_metadata='NO_LOG';
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo NO_LOG Expected
+
+SET @@GLOBAL.binlog_row_metadata='MINIMAL';
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo MINIMAL Expected
+
+SET @@GLOBAL.binlog_row_metadata='FULL';
+SELECT @@GLOBAL.binlog_row_metadata;
+--echo FULL Expected
+
+--echo '#---------------------BS_STVARS_002_03----------------------#'
+#################################################################
+# Check if the value in GLOBAL Table matches value in variable  #
+#################################################################
+
+SET @@GLOBAL.binlog_row_metadata='MINIMAl';
+SELECT *
+FROM information_schema.global_variables
+WHERE VARIABLE_NAME='binlog_row_metadata';
+
+--echo '#---------------------BS_STVARS_002_04----------------------#'
+#################################################################
+# Check if the value in SESSION Table matches value in variable #
+#################################################################
+
+SELECT *
+FROM information_schema.session_variables
+WHERE VARIABLE_NAME LIKE 'binlog_row_metadata';
+
+--echo '#---------------------BS_STVARS_002_05----------------------#'
+################################################################################
+#   Check if binlog_row_metadata can be accessed with and without @@ sign      #
+################################################################################
+
+SELECT COUNT(@@binlog_row_metadata);
+--echo 1 Expected
+SELECT COUNT(@@GLOBAL.binlog_row_metadata);
+--echo 1 Expected
+
+--echo '#---------------------BS_STVARS_002_06----------------------#'
+################################################################################
+#   Check if binlog_row_metadata can handle invalid values correctly           #
+################################################################################
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL binlog_row_metadata = full1;
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL binlog_row_metadata = "full1";
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL binlog_row_metadata = 3;
+
+--error ER_WRONG_VALUE_FOR_VAR
+SET GLOBAL binlog_row_metadata = -1;
+
+SET @@global.binlog_row_metadata= @start_value;

sql/field.cc

@@ -3493,11 +3493,12 @@ void Field_new_decimal::sql_type(String &str) const
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_new_decimal::save_field_metadata(uchar *metadata_ptr)
+
+Binlog_type_info Field_new_decimal::binlog_type_info() const
 {
-  *metadata_ptr= precision;
-  *(metadata_ptr + 1)= decimals();
-  return 2;
+  DBUG_ASSERT(Field_new_decimal::type() == binlog_type());
+  return Binlog_type_info(Field_new_decimal::type(), precision +
+                          (decimals() << 8), 2, binlog_signess());
 }
 
 
@@ -4665,10 +4666,11 @@ bool Field_float::send_binary(Protocol *protocol)
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_float::save_field_metadata(uchar *metadata_ptr)
+Binlog_type_info Field_float::binlog_type_info() const
 {
-  *metadata_ptr= pack_length();
-  return 1;
+  DBUG_ASSERT(Field_float::type() == binlog_type());
+  return Binlog_type_info(Field_float::type(), pack_length(), 1,
+                          binlog_signess());
 }
 
 
@@ -4976,10 +4978,11 @@ void Field_double::sort_string(uchar *to,uint length __attribute__((unused)))
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_double::save_field_metadata(uchar *metadata_ptr)
+Binlog_type_info Field_double::binlog_type_info() const
 {
-  *metadata_ptr= pack_length();
-  return 1;
+  DBUG_ASSERT(Field_double::type() == binlog_type());
+  return Binlog_type_info(Field_double::type(), pack_length(), 1,
+                          binlog_signess());
 }
 
 
@@ -5627,6 +5630,11 @@ bool Field_timestampf::val_native(Native *to)
   return Field::val_native(to);
 }
 
+Binlog_type_info Field_timestampf::binlog_type_info() const
+{
+  return Binlog_type_info(Field_timestampf::binlog_type(), decimals(), 1);
+}
+
 
 /*************************************************************/
 sql_mode_t Field_temporal::can_handle_sql_mode_dependency_on_store() const
@@ -6244,6 +6252,10 @@ bool Field_timef::get_date(MYSQL_TIME *ltime, date_mode_t fuzzydate)
   TIME_from_longlong_time_packed(ltime, tmp);
   return false;
 }
+Binlog_type_info Field_timef::binlog_type_info() const
+{
+  return Binlog_type_info(Field_timef::binlog_type(), decimals(), 1);
+}
 
 /****************************************************************************
 ** year type
@@ -6955,6 +6967,10 @@ bool Field_datetimef::get_TIME(MYSQL_TIME *ltime, const uchar *pos,
   TIME_from_longlong_datetime_packed(ltime, tmp);
   return validate_MMDD(tmp, ltime->month, ltime->day, fuzzydate);
 }
+Binlog_type_info Field_datetimef::binlog_type_info() const
+{
+  return Binlog_type_info(Field_datetimef::binlog_type(), decimals(), 1);
+}
 
 /****************************************************************************
 ** string type
@@ -7536,15 +7552,16 @@ Field_string::unpack(uchar *to, const uchar *from, const uchar *from_end,
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_string::save_field_metadata(uchar *metadata_ptr)
+Binlog_type_info Field_string::binlog_type_info() const
 {
+  uint16 a;
   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;
+                     field_length, real_type()));
+  a= (real_type() ^ ((field_length & 0x300) >> 4)) + (((uint)(field_length & 0xFF)) << 8);
+  DBUG_ASSERT(Field_string::type() == binlog_type());
+  return Binlog_type_info(Field_string::type(), a, 2, charset());
 }
 
 
@@ -7633,11 +7650,10 @@ const uint Field_varstring::MAX_SIZE= UINT_MAX16;
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_varstring::save_field_metadata(uchar *metadata_ptr)
+Binlog_type_info Field_varstring::binlog_type_info() const
 {
-  DBUG_ASSERT(field_length <= 65535);
-  int2store((char*)metadata_ptr, field_length);
-  return 2;
+  DBUG_ASSERT(Field_varstring::type() == binlog_type());
+  return Binlog_type_info(Field_varstring::type(), field_length, 2, charset());
 }
 
 
@@ -8270,6 +8286,11 @@ int Field_varstring_compressed::cmp_max(const uchar *a_ptr, const uchar *b_ptr,
 
   return sortcmp(&a, &b, field_charset());
 }
+Binlog_type_info Field_varstring_compressed::binlog_type_info() const
+{
+  return Binlog_type_info(Field_varstring_compressed::binlog_type(),
+                          field_length, 2, charset());
+}
 
 
 /****************************************************************************
@@ -8614,12 +8635,11 @@ Field *Field_blob::new_key_field(MEM_ROOT *root, TABLE *new_table,
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_blob::save_field_metadata(uchar *metadata_ptr)
+Binlog_type_info Field_blob::binlog_type_info() const
 {
-  DBUG_ENTER("Field_blob::save_field_metadata");
-  *metadata_ptr= pack_length_no_ptr();
-  DBUG_PRINT("debug", ("metadata: %u (pack_length_no_ptr)", *metadata_ptr));
-  DBUG_RETURN(1);
+  DBUG_ASSERT(Field_blob::type() == binlog_type());
+  return Binlog_type_info(Field_blob::type(), pack_length_no_ptr(), 1,
+                          charset());
 }
 
 
@@ -8874,6 +8894,11 @@ longlong Field_blob_compressed::val_int(void)
                                       buf.ptr(), buf.length()).result();
 }
 
+Binlog_type_info Field_blob_compressed::binlog_type_info() const
+{
+  return Binlog_type_info(Field_blob_compressed::binlog_type(),
+                          pack_length_no_ptr(), 1, charset());
+}
 
 /****************************************************************************
 ** enum type.
@@ -9008,11 +9033,11 @@ longlong Field_enum::val_int(const uchar *real_ptr) const
 
    @returns number of bytes written to metadata_ptr
 */
-int Field_enum::save_field_metadata(uchar *metadata_ptr)
+Binlog_type_info Field_enum::binlog_type_info() const
 {
-  *metadata_ptr= real_type();
-  *(metadata_ptr + 1)= pack_length();
-  return 2;
+  DBUG_ASSERT(Field_enum::type() == binlog_type());
+  return Binlog_type_info(Field_enum::type(), real_type() + (pack_length() << 8),
+                          2, charset(), (TYPELIB *)get_typelib(), NULL);
 }
 
 
@@ -9215,6 +9240,13 @@ void Field_set::sql_type(String &res) const
   res.append(')');
 }
 
+Binlog_type_info Field_set::binlog_type_info() const
+{
+  DBUG_ASSERT(Field_set::type() == binlog_type());
+  return Binlog_type_info(Field_set::type(), real_type()
+           + (pack_length() << 8), 2, charset(), NULL, (TYPELIB *)get_typelib());
+}
+
 /**
   @retval
     1  if the fields are equally defined
@@ -9742,33 +9774,6 @@ uint Field_bit::get_key_image(uchar *buff, uint length, imagetype type_arg)
 }
 
 
-/**
-   Save the field metadata for bit fields.
-
-   Saves the bit length in the first byte and bytes in record in the
-   second byte of the field metadata array at index of *metadata_ptr and
-   *(metadata_ptr + 1).
-
-   @param   metadata_ptr   First byte of field metadata
-
-   @returns number of bytes written to metadata_ptr
-*/
-int Field_bit::save_field_metadata(uchar *metadata_ptr)
-{
-  DBUG_ENTER("Field_bit::save_field_metadata");
-  DBUG_PRINT("debug", ("bit_len: %d, bytes_in_rec: %d",
-                       bit_len, bytes_in_rec));
-  /*
-    Since this class and Field_bit_as_char have different ideas of
-    what should be stored here, we compute the values of the metadata
-    explicitly using the field_length.
-   */
-  metadata_ptr[0]= field_length % 8;
-  metadata_ptr[1]= field_length / 8;
-  DBUG_RETURN(2);
-}
-
-
 /**
    Returns the number of bytes field uses in row-based replication 
    row packed size.

sql/log_event.cc

@@ -3428,7 +3428,8 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len,
     m_colcnt(0), m_coltype(0),
     m_memory(NULL), m_table_id(ULONGLONG_MAX), m_flags(0),
     m_data_size(0), m_field_metadata(0), m_field_metadata_size(0),
-    m_null_bits(0), m_meta_memory(NULL)
+    m_null_bits(0), m_meta_memory(NULL),
+    m_optional_metadata_len(0), m_optional_metadata(NULL)
 {
   unsigned int bytes_read= 0;
   DBUG_ENTER("Table_map_log_event::Table_map_log_event(const char*,uint,...)");
@@ -3517,8 +3518,28 @@ Table_map_log_event::Table_map_log_event(const char *buf, uint event_len,
       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);
+      ptr_after_colcnt= (unsigned char*)ptr_after_colcnt + num_null_bytes;
+    }
+
+    bytes_read= (uint) (ptr_after_colcnt - (uchar *)buf);
+
+    /* After null_bits field, there are some new fields for extra metadata. */
+    if (bytes_read < event_len)
+    {
+      m_optional_metadata_len= event_len - bytes_read;
+      m_optional_metadata=
+        static_cast<unsigned char*>(my_malloc(m_optional_metadata_len, MYF(MY_WME)));
+      memcpy(m_optional_metadata, ptr_after_colcnt, m_optional_metadata_len);
     }
   }
+#ifdef MYSQL_SERVER
+  if (!m_table)
+    DBUG_VOID_RETURN;
+  binlog_type_info_array= (Binlog_type_info *)thd->alloc(m_table->s->fields *
+                                                         sizeof(Binlog_type_info));
+  for (uint i= 0; i <  m_table->s->fields; i++)
+    binlog_type_info_array[i]= m_table->field[i]->binlog_type_info();
+#endif
 
   DBUG_VOID_RETURN;
 }
@@ -3528,6 +3549,237 @@ Table_map_log_event::~Table_map_log_event()
 {
   my_free(m_meta_memory);
   my_free(m_memory);
+  my_free(m_optional_metadata);
+  m_optional_metadata= NULL;
+}
+
+/**
+   Parses SIGNEDNESS field.
+
+   @param[out] vec     stores the signedness flags extracted from field.
+   @param[in]  field   SIGNEDNESS field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_signedness(std::vector<bool> &vec,
+                             unsigned char *field, unsigned int length)
+{
+  for (unsigned int i= 0; i < length; i++)
+  {
+    for (unsigned char c= 0x80; c != 0; c>>= 1)
+      vec.push_back(field[i] & c);
+  }
+}
+
+/**
+   Parses DEFAULT_CHARSET field.
+
+   @param[out] default_charset  stores collation numbers extracted from field.
+   @param[in]  field   DEFAULT_CHARSET field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_default_charset(Table_map_log_event::Optional_metadata_fields::
+                                  Default_charset &default_charset,
+                                  unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  default_charset.default_charset= net_field_length(&p);
+  while (p < field + length)
+  {
+    unsigned int col_index= net_field_length(&p);
+    unsigned int col_charset= net_field_length(&p);
+
+    default_charset.charset_pairs.push_back(std::make_pair(col_index,
+                                                           col_charset));
+  }
+}
+
+/**
+   Parses COLUMN_CHARSET field.
+
+   @param[out] vec     stores collation numbers extracted from field.
+   @param[in]  field   COLUMN_CHARSET field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_column_charset(std::vector<unsigned int> &vec,
+                                 unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  while (p < field + length)
+    vec.push_back(net_field_length(&p));
+}
+
+/**
+   Parses COLUMN_NAME field.
+
+   @param[out] vec     stores column names extracted from field.
+   @param[in]  field   COLUMN_NAME field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_column_name(std::vector<std::string> &vec,
+                              unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  while (p < field + length)
+  {
+    unsigned len= net_field_length(&p);
+    vec.push_back(std::string(reinterpret_cast<char *>(p), len));
+    p+= len;
+  }
+}
+
+/**
+   Parses SET_STR_VALUE/ENUM_STR_VALUE field.
+
+   @param[out] vec     stores SET/ENUM column's string values extracted from
+                       field. Each SET/ENUM column's string values are stored
+                       into a string separate vector. All of them are stored
+                       in 'vec'.
+   @param[in]  field   COLUMN_NAME field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_set_str_value(std::vector<Table_map_log_event::
+                                Optional_metadata_fields::str_vector> &vec,
+                                unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  while (p < field + length)
+  {
+    unsigned int count= net_field_length(&p);
+
+    vec.push_back(std::vector<std::string>());
+    for (unsigned int i= 0; i < count; i++)
+    {
+      unsigned len1= net_field_length(&p);
+      vec.back().push_back(std::string(reinterpret_cast<char *>(p), len1));
+      p+= len1;
+    }
+  }
+}
+
+/**
+   Parses GEOMETRY_TYPE field.
+
+   @param[out] vec     stores geometry column's types extracted from field.
+   @param[in]  field   GEOMETRY_TYPE field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_geometry_type(std::vector<unsigned int> &vec,
+                                unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  while (p < field + length)
+    vec.push_back(net_field_length(&p));
+}
+
+/**
+   Parses SIMPLE_PRIMARY_KEY field.
+
+   @param[out] vec     stores primary key's column information extracted from
+                       field. Each column has an index and a prefix which are
+                       stored as a unit_pair. prefix is always 0 for
+                       SIMPLE_PRIMARY_KEY field.
+   @param[in]  field   SIMPLE_PRIMARY_KEY field in table_map_event.
+   @param[in]  length  length of the field
+ */
+static void parse_simple_pk(std::vector<Table_map_log_event::
+                            Optional_metadata_fields::uint_pair> &vec,
+                            unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  while (p < field + length)
+    vec.push_back(std::make_pair(net_field_length(&p), 0));
+}
+
+/**
+   Parses PRIMARY_KEY_WITH_PREFIX field.
+
+   @param[out] vec     stores primary key's column information extracted from
+                       field. Each column has an index and a prefix which are
+                       stored as a unit_pair.
+   @param[in]  field   PRIMARY_KEY_WITH_PREFIX field in table_map_event.
+   @param[in]  length  length of the field
+ */
+
+static void parse_pk_with_prefix(std::vector<Table_map_log_event::
+                                 Optional_metadata_fields::uint_pair> &vec,
+                                 unsigned char *field, unsigned int length)
+{
+  unsigned char* p= field;
+
+  while (p < field + length)
+  {
+    unsigned int col_index= net_field_length(&p);
+    unsigned int col_prefix= net_field_length(&p);
+    vec.push_back(std::make_pair(col_index, col_prefix));
+  }
+}
+
+Table_map_log_event::Optional_metadata_fields::
+Optional_metadata_fields(unsigned char* optional_metadata,
+                         unsigned int optional_metadata_len)
+{
+  unsigned char* field= optional_metadata;
+
+  if (optional_metadata == NULL)
+    return;
+
+  while (field < optional_metadata + optional_metadata_len)
+  {
+    unsigned int len;
+    Optional_metadata_field_type type=
+      static_cast<Optional_metadata_field_type>(field[0]);
+
+    // Get length and move field to the value.
+    field++;
+    len= net_field_length(&field);
+
+    switch(type)
+    {
+    case SIGNEDNESS:
+      parse_signedness(m_signedness, field, len);
+      break;
+    case DEFAULT_CHARSET:
+      parse_default_charset(m_default_charset, field, len);
+      break;
+    case COLUMN_CHARSET:
+      parse_column_charset(m_column_charset, field, len);
+      break;
+    case COLUMN_NAME:
+      parse_column_name(m_column_name, field, len);
+      break;
+    case SET_STR_VALUE:
+      parse_set_str_value(m_set_str_value, field, len);
+      break;
+    case ENUM_STR_VALUE:
+      parse_set_str_value(m_enum_str_value, field, len);
+      break;
+    case GEOMETRY_TYPE:
+      parse_geometry_type(m_geometry_type, field, len);
+      break;
+    case SIMPLE_PRIMARY_KEY:
+      parse_simple_pk(m_primary_key, field, len);
+      break;
+    case PRIMARY_KEY_WITH_PREFIX:
+      parse_pk_with_prefix(m_primary_key, field, len);
+      break;
+    case ENUM_AND_SET_DEFAULT_CHARSET:
+      parse_default_charset(m_enum_and_set_default_charset, field, len);
+      break;
+    case ENUM_AND_SET_COLUMN_CHARSET:
+      parse_column_charset(m_enum_and_set_column_charset, field, len);
+      break;
+    default:
+      DBUG_ASSERT(0);
+    }
+    // next field
+    field+= len;
+  }
 }
 
 

sql/mysqld.cc

@@ -450,6 +450,7 @@ my_bool opt_noacl;
 my_bool sp_automatic_privileges= 1;
 
 ulong opt_binlog_rows_event_max_size;
+ulong binlog_row_metadata;
 my_bool opt_master_verify_checksum= 0;
 my_bool opt_slave_sql_verify_checksum= 1;
 const char *binlog_format_names[]= {"MIXED", "STATEMENT", "ROW", NullS};

sql/mysqld.h

@@ -245,6 +245,7 @@ extern ulonglong max_binlog_cache_size, max_binlog_stmt_cache_size;
 extern ulong max_binlog_size;
 extern ulong slave_max_allowed_packet;
 extern ulong opt_binlog_rows_event_max_size;
+extern ulong binlog_row_metadata;
 extern ulong thread_cache_size;
 extern ulong stored_program_cache_size;
 extern ulong opt_slave_parallel_threads;

sql/sql_type_geom.cc

@@ -912,4 +912,11 @@ uint Field_geom::get_key_image(uchar *buff,uint length, imagetype type_arg)
   return Field_blob::get_key_image_itRAW(buff, length);
 }
 
+Binlog_type_info Field_geom::binlog_type_info() const
+{
+  DBUG_ASSERT(Field_geom::type() == binlog_type());
+  return Binlog_type_info(Field_geom::type(), pack_length_no_ptr(), 1,
+                          field_charset(), type_handler_geom()->geometry_type());
+}
+
 #endif // HAVE_SPATIAL

sql/sql_type_geom.h

@@ -435,6 +435,7 @@ class Field_geom :public Field_blob
   {
     out->append(STRING_WITH_LEN("unprintable_geometry_value"));
   }
+  Binlog_type_info binlog_type_info()  const;
 };
 
 #endif // HAVE_SPATIAL

sql/sys_vars.cc

@@ -6209,6 +6209,19 @@ static Sys_var_enum Sys_binlog_row_image(
        SESSION_VAR(binlog_row_image), CMD_LINE(REQUIRED_ARG),
        binlog_row_image_names, DEFAULT(BINLOG_ROW_IMAGE_FULL));
 
+static const char *binlog_row_metadata_names[]= {"NO_LOG", "MINIMAL", "FULL", NullS};
+static Sys_var_enum Sys_binlog_row_metadata(
+       "binlog_row_metadata",
+       "Controls whether metadata is logged using FULL , MINIMAL format and NO_LOG."
+       "FULL causes all metadata to be logged; MINIMAL means that only "
+       "metadata actually required by slave is logged; NO_LOG NO metadata will be logged."
+       "Default: NO_LOG.",
+       GLOBAL_VAR(binlog_row_metadata), CMD_LINE(REQUIRED_ARG),
+       binlog_row_metadata_names, DEFAULT(Table_map_log_event::BINLOG_ROW_METADATA_NO_LOG),
+       NO_MUTEX_GUARD, NOT_IN_BINLOG, ON_CHECK(NULL),
+       ON_UPDATE(NULL));
+
+
 static bool check_pseudo_slave_mode(sys_var *self, THD *thd, set_var *var)
 {
   longlong previous_val= thd->variables.pseudo_slave_mode;