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.

1 files changed, 253 insertions, 1 deletions

diff --git a/sql/log_event.cc b/sql/log_event.cc
index 351caa96446..560fec81f9e 100644
--- a/sql/log_event.cc
+++ b/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;
+  }
 }