Add an event to send column information in RBR

Although Table_map events include some column metadata information,
such as type and length, it lacks the necessary information to fully
deduce column's definition and to interpret its values. For example,
it lacks information such as the column name, whether an integer type
is signed or unsigned, or the character set of string types. This
makes it rather difficult for external programs to extract meaningful
row data from row-based events.

In order to enable external programs to fully deduce the table/column
definitions, a new Table_metadata event is added. The purpose of this
event is to describe the structure and other properties of the table
and its columns, such as name, SQL type name, character set, etc. Each
Table_metadata event follows a Table_map. Whether Table_metadata
events are written to the binary log can be controlled using the
system variable binlog_rows_table_metadata_events.

Author: Davi Arnaut

client/mysqlbinlog.cc

@@ -669,6 +669,37 @@ write_event_header_and_base64(Log_event *ev, FILE *result_file,
 }
 
 
+/**
+  Attach a Table_metadata event to its respective Table_map event;
+
+  @param event The Table_metadata event.
+  @param print_event_info Print context.
+
+  @remark Upon successful attachment, the Table_map takes ownership
+          over the Table_metadata event.
+
+  @return false if successful, true otherwise.
+*/
+static bool
+attach_to_table_map_event(Table_metadata_log_event *event,
+                          PRINT_EVENT_INFO *print_event_info)
+{
+  Table_map_log_event *table_map;
+  ulong table_map_id= event->get_table_id();
+  DBUG_ENTER("process_table_metadata_event");
+
+  if (!(table_map= print_event_info->m_table_map.get_table(table_map_id)))
+    table_map= print_event_info->m_table_map_ignored.get_table(table_map_id);
+
+  if (!table_map)
+    DBUG_RETURN(true);
+
+  table_map->set_table_metadata_event(event);
+
+  DBUG_RETURN(false);
+}
+
+
 /**
   Print the given event, and either delete it or delegate the deletion
   to someone else.
@@ -903,6 +934,13 @@ Exit_status process_event(PRINT_EVENT_INFO *print_event_info, Log_event *ev,
 	my_free(fname);
       break;
     }
+    case TABLE_METADATA_EVENT:
+    {
+      Table_metadata_log_event *event= (Table_metadata_log_event *) ev;
+      destroy_evt= attach_to_table_map_event(event, print_event_info);
+      event->print(result_file, print_event_info);
+      break;
+    }
     case TABLE_MAP_EVENT:
     {
       Table_map_log_event *map= ((Table_map_log_event *)ev);

mysql-test/include/show_binlog_events2.inc

@@ -1,4 +1,4 @@
---let $binlog_start=107
+--let $binlog_start=130
 --replace_result $binlog_start <binlog_start>
 --replace_column 2 # 5 #
 --replace_regex /\/\* xid=.* \*\//\/* XID *\// /table_id: [0-9]+/table_id: #/