diff --git a/sql/log_event.cc b/sql/log_event.cc
index c814b65066cdd..5e5e99dcd0948 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -17,6 +17,7 @@
 
 
 #include "mariadb.h"
+#include "mysql/psi/psi_base.h"
 #include "sql_priv.h"
 #include "handler.h"
 #ifndef MYSQL_CLIENT
@@ -3718,17 +3719,32 @@ Table_map_log_event::~Table_map_log_event()
 /**
    Parses SIGNEDNESS field.
 
-   @param[out] vec     stores the signedness flags extracted from field.
+   @param[out] column_metadata store all column metadata here (including signedness)
    @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)
+static void parse_signedness(
+  Dynamic_array<Table_map_log_event::Optional_metadata_fields::Column_metadata>& column_metadata,
+  unsigned char *field, unsigned int length)
 {
-  for (unsigned int i= 0; i < length; i++)
+  unsigned char *field_end= field + length;
+  unsigned char unsigned_bitfield= 0;
+  unsigned char mask= 0;
+
+  for (uint col= 0; col < column_metadata.size(); col++)
   {
-    for (unsigned char c= 0x80; c != 0; c>>= 1)
-      vec.push_back(field[i] & c);
+    auto &col_meta= column_metadata.at(col);
+    if (!is_numeric_type(col_meta.column_type))
+      continue;
+    if (mask == 0)
+    {
+      if (field >= field_end)
+        return;
+      unsigned_bitfield= *field++;
+      mask= 0x80;
+    }
+    col_meta.is_unsigned= unsigned_bitfield & mask;
+    mask >>= 1;
   }
 }
 
@@ -3751,8 +3767,8 @@ static void parse_default_charset(Table_map_log_event::Optional_metadata_fields:
     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));
+    default_charset.charset_pairs.append(std::make_pair(col_index,
+                                                        col_charset));
   }
 }
 
@@ -3763,36 +3779,38 @@ static void parse_default_charset(Table_map_log_event::Optional_metadata_fields:
    @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,
+static void parse_column_charset(Dynamic_array<uint> &vec,
                                  unsigned char *field, unsigned int length)
 {
   unsigned char* p= field;
 
   while (p < field + length)
-    vec.push_back(net_field_length(&p));
+    vec.append(net_field_length(&p));
 }
 
 /**
    Parses COLUMN_NAME field.
 
    @param[in]  root    Allocate memory here
-   @param[out] name    Store column names extracted from field here
+   @param[out] column_metadata store all column metadata here (including column names)
    @param[in]  field   COLUMN_NAME field in table_map_event.
    @param[in]  length  length of the field
  */
-static bool parse_column_name(MEM_ROOT *root, LEX_CSTRING *name,
-                              unsigned char *field, unsigned int length)
+static bool parse_column_name(MEM_ROOT *root, 
+  Dynamic_array<Table_map_log_event::Optional_metadata_fields::Column_metadata>& column_metadata,
+  unsigned char *field, unsigned int length)
 {
-  for (uchar *end= field+length; field < end ; name++)
+  unsigned int col = 0;
+  for (uchar *end= field+length; field < end && col < column_metadata.size(); col++)
   {
+    LEX_CSTRING& name = column_metadata.at(col).column_name;
     uint name_length= net_field_length(&field);
-    if (!(name->str= strmake_root(root, (char*) field, name_length)))
+    if (!(name.str= strmake_root(root, (char*) field, name_length)))
       return 1;
-    name->length= name_length;
+    name.length= name_length;
     field+= name_length;
   }
 
-  name->str= 0;                                 // End marker
   return 0;
 }
 
@@ -3806,9 +3824,10 @@ static bool parse_column_name(MEM_ROOT *root, LEX_CSTRING *name,
    @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)
+static void parse_set_str_value(
+    Dynamic_array<Table_map_log_event::Optional_metadata_fields::str_vector>
+        &vec,
+    unsigned char *field, unsigned int length)
 {
   unsigned char* p= field;
 
@@ -3816,11 +3835,14 @@ static void parse_set_str_value(std::vector<Table_map_log_event::
   {
     unsigned int count= net_field_length(&p);
 
-    vec.push_back(std::vector<std::string>());
+    if (vec.reserve(vec.elements() + 1))
+      return;
+    vec.elements(vec.elements() + 1);
+    vec.back()->init(PSI_INSTRUMENT_MEM);
     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));
+      vec.back()->append(LEX_CSTRING{reinterpret_cast<char *>(p), len1});
       p+= len1;
     }
   }
@@ -3833,13 +3855,13 @@ static void parse_set_str_value(std::vector<Table_map_log_event::
    @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,
+static void parse_geometry_type(Dynamic_array<uint> &vec,
                                 unsigned char *field, unsigned int length)
 {
   unsigned char* p= field;
 
   while (p < field + length)
-    vec.push_back(net_field_length(&p));
+    vec.append(net_field_length(&p));
 }
 
 /**
@@ -3852,14 +3874,15 @@ static void parse_geometry_type(std::vector<unsigned int> &vec,
    @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)
+static void parse_simple_pk(
+    Dynamic_array<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));
+    vec.append(std::make_pair(net_field_length(&p), 0));
 }
 
 /**
@@ -3872,9 +3895,10 @@ static void parse_simple_pk(std::vector<Table_map_log_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)
+static void parse_pk_with_prefix(
+    Dynamic_array<Table_map_log_event::Optional_metadata_fields::uint_pair>
+        &vec,
+    unsigned char *field, unsigned int length)
 {
   unsigned char* p= field;
 
@@ -3882,21 +3906,36 @@ static void parse_pk_with_prefix(std::vector<Table_map_log_event::
   {
     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));
+    vec.append(std::make_pair(col_index, col_prefix));
   }
 }
 
 Table_map_log_event::Optional_metadata_fields::
 Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
+                         const uchar* column_types,
                          uchar* optional_metadata,
                          size_t optional_metadata_len,
                          bool only_column_names)
+    : m_column_charset(PSI_INSTRUMENT_MEM),
+      m_enum_and_set_column_charset(PSI_INSTRUMENT_MEM),
+      m_enum_str_value(PSI_INSTRUMENT_MEM),
+      m_set_str_value(PSI_INSTRUMENT_MEM), m_geometry_type(PSI_INSTRUMENT_MEM),
+      m_primary_key(PSI_INSTRUMENT_MEM),
+      m_column_metadata(PSI_INSTRUMENT_MEM, master_columns)
 {
   unsigned int len;
   uchar *metadata_end;
 
   allocation_error= 0;
-  m_column_name= 0;
+
+  for (uint i= 0; i < master_columns; i++)
+  {
+    Column_metadata column_metadata{};
+    column_metadata.column_type = column_types[i];
+    if (m_column_metadata.append(column_metadata))
+      goto error;
+  }
+
   if (optional_metadata == NULL)
     return;
 
@@ -3914,7 +3953,7 @@ Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
 
     switch(type) {
     case SIGNEDNESS:
-      parse_signedness(m_signedness, field, len);
+      parse_signedness(m_column_metadata, field, len);
       break;
     case DEFAULT_CHARSET:
       parse_default_charset(m_default_charset, field, len);
@@ -3923,11 +3962,7 @@ Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
       parse_column_charset(m_column_charset, field, len);
       break;
     case COLUMN_NAME:
-      if (!(m_column_name= (LEX_CSTRING*) alloc_root(root,
-                                                     sizeof(LEX_CSTRING) *
-                                                     (master_columns +1 ))))
-        goto error;
-      if (parse_column_name(root, m_column_name, field, len))
+      if (parse_column_name(root, m_column_metadata, field, len))
         goto error;
       break;
     case SET_STR_VALUE:
diff --git a/sql/log_event.h b/sql/log_event.h
index 629ffc23476eb..997f6f7b53680 100644
--- a/sql/log_event.h
+++ b/sql/log_event.h
@@ -31,6 +31,7 @@
 
 #include <my_bitmap.h>
 #include "rpl_constants.h"
+#include "sql_array.h"
 #include <vector>
 #include <string>
 #include <functional>
@@ -38,6 +39,24 @@
 #include <map>
 #include <lex_charset.h>
 
+static inline bool is_numeric_type(uint type)
+{
+  switch (type)
+  {
+  case MYSQL_TYPE_TINY:
+  case MYSQL_TYPE_SHORT:
+  case MYSQL_TYPE_INT24:
+  case MYSQL_TYPE_LONG:
+  case MYSQL_TYPE_LONGLONG:
+  case MYSQL_TYPE_NEWDECIMAL:
+  case MYSQL_TYPE_FLOAT:
+  case MYSQL_TYPE_DOUBLE:
+    return true;
+  default:
+    return false;
+  }
+}
+
 #ifdef MYSQL_CLIENT
 #include "sql_const.h"
 #include "rpl_utility.h"
@@ -4502,19 +4521,22 @@ class Table_map_log_event : public Log_event
   struct Optional_metadata_fields
   {
     typedef std::pair<unsigned int, unsigned int> uint_pair;
-    typedef std::vector<std::string> str_vector;
+    typedef Dynamic_array<LEX_CSTRING> str_vector;
     bool allocation_error; /* Set if allocation of data structures fails */
 
     struct Default_charset
     {
-      Default_charset() : default_charset(0) {}
+      Default_charset() : default_charset(0), charset_pairs(PSI_INSTRUMENT_MEM)
+      {
+      }
       bool empty() const { return default_charset == 0; }
 
       // Default charset for the columns which are not in charset_pairs.
       unsigned int default_charset;
 
       /* The uint_pair means <column index, column charset number>. */
-      std::vector<uint_pair> charset_pairs;
+      // std::vector<uint_pair> charset_pairs;
+      Dynamic_array<uint_pair> charset_pairs;
     };
 
     // Contents of DEFAULT_CHARSET field is converted into Default_charset.
@@ -4522,21 +4544,28 @@ class Table_map_log_event : public Log_event
     // Contents of ENUM_AND_SET_DEFAULT_CHARSET are converted into
     // Default_charset.
     Default_charset m_enum_and_set_default_charset;
-    std::vector<bool> m_signedness;
     // Character set number of every string column
-    std::vector<unsigned int> m_column_charset;
+    Dynamic_array<uint> m_column_charset;
     // Character set number of every ENUM or SET column.
-    std::vector<unsigned int> m_enum_and_set_column_charset;
-    LEX_CSTRING *m_column_name;
+    Dynamic_array<uint> m_enum_and_set_column_charset;
     // each str_vector stores values of one enum/set column
-    std::vector<str_vector> m_enum_str_value;
-    std::vector<str_vector> m_set_str_value;
-    std::vector<unsigned int> m_geometry_type;
+    Dynamic_array<str_vector> m_enum_str_value;
+    Dynamic_array<str_vector> m_set_str_value;
+    Dynamic_array<uint> m_geometry_type;
     /*
       The uint_pair means <column index, prefix length>.  Prefix length is 0 if
       whole column value is used.
     */
-    std::vector<uint_pair> m_primary_key;
+    Dynamic_array<uint_pair> m_primary_key;
+
+    struct Column_metadata
+    {
+        LEX_CSTRING column_name{ nullptr, 0 };
+        uchar column_type{};
+        bool is_unsigned{ false };
+    };
+
+    Dynamic_array<Column_metadata> m_column_metadata;
 
     /*
       It parses m_optional_metadata and populates into above variables.
@@ -4549,6 +4578,7 @@ class Table_map_log_event : public Log_event
       @param[in] only_column_names Only read column names
      */
     Optional_metadata_fields(MEM_ROOT *root, uint master_cols,
+                             const uchar* column_types,
                              uchar* optional_metadata,
                              size_t optional_metadata_len,
                              bool only_column_names);
diff --git a/sql/log_event_client.cc b/sql/log_event_client.cc
index 468c196d8b6f2..08297012b1183 100644
--- a/sql/log_event_client.cc
+++ b/sql/log_event_client.cc
@@ -286,24 +286,6 @@ static bool hexdump_data_to_io_cache(IO_CACHE *file,
   return 1;
 }
 
-static inline bool is_numeric_type(uint type)
-{
-  switch (type)
-  {
-  case MYSQL_TYPE_TINY:
-  case MYSQL_TYPE_SHORT:
-  case MYSQL_TYPE_INT24:
-  case MYSQL_TYPE_LONG:
-  case MYSQL_TYPE_LONGLONG:
-  case MYSQL_TYPE_NEWDECIMAL:
-  case MYSQL_TYPE_FLOAT:
-  case MYSQL_TYPE_DOUBLE:
-    return true;
-  default:
-    return false;
-  }
-  return false;
-}
 
 static inline bool is_character_type(uint type)
 {
@@ -3174,7 +3156,9 @@ bool Table_map_log_event::print(FILE *file, PRINT_EVENT_INFO *print_event_info)
     {
       MEM_ROOT root;
       init_alloc_root(0, &root, 4096, 0, 0);
-      Optional_metadata_fields fields(&root, m_colcnt,
+      Optional_metadata_fields fields(&root, 
+                                      m_colcnt,
+                                      m_coltype,
                                       m_optional_metadata,
                                       m_optional_metadata_len,
                                       0);
@@ -3230,7 +3214,7 @@ class Table_map_log_event::Charset_iterator
   */
   static std::unique_ptr<Charset_iterator> create_charset_iterator(
       const Default_charset &default_charset,
-      const std::vector<uint> &column_charset);
+      const Dynamic_array<uint> &column_charset);
 };
 
 /**
@@ -3240,9 +3224,8 @@ class Table_map_log_event::Default_charset_iterator : public Charset_iterator
 {
  public:
   Default_charset_iterator(const Default_charset &default_charset)
-      : m_iterator(default_charset.charset_pairs.begin()),
-        m_end(default_charset.charset_pairs.end()),
-        m_column_index(0),
+      : m_iterator(default_charset.charset_pairs.front()),
+        m_end(default_charset.charset_pairs.end()), m_column_index(0),
         m_default_charset_info(
             get_charset(default_charset.default_charset, 0)) {}
 
@@ -3259,8 +3242,8 @@ class Table_map_log_event::Default_charset_iterator : public Charset_iterator
   ~Default_charset_iterator(){};
 
  private:
-  std::vector<Optional_metadata_fields::uint_pair>::const_iterator m_iterator,
-      m_end;
+  const Optional_metadata_fields::uint_pair *m_iterator;
+  const Optional_metadata_fields::uint_pair *m_end;
   uint m_column_index;
   const CHARSET_INFO *m_default_charset_info;
 };
@@ -3271,8 +3254,9 @@ class Table_map_log_event::Default_charset_iterator : public Charset_iterator
 class Table_map_log_event::Column_charset_iterator : public Charset_iterator
 {
  public:
-  Column_charset_iterator(const std::vector<uint> &column_charset)
-      : m_iterator(column_charset.begin()), m_end(column_charset.end()) {}
+  Column_charset_iterator(const Dynamic_array<uint> &column_charset)
+      : m_iterator(column_charset.front()), m_end(column_charset.end())
+  {}
 
   const CHARSET_INFO *next() override {
     const CHARSET_INFO *ret = nullptr;
@@ -3285,15 +3269,15 @@ class Table_map_log_event::Column_charset_iterator : public Charset_iterator
 
  ~Column_charset_iterator(){};
  private:
-  std::vector<uint>::const_iterator m_iterator;
-  std::vector<uint>::const_iterator m_end;
+  const uint *m_iterator;
+  const uint *m_end;
 };
 //Table_map_log_event::Column_charset_iterator::~Column_charset_iterator(){int a=8;a++; a--;};
 
 std::unique_ptr<Table_map_log_event::Charset_iterator>
 Table_map_log_event::Charset_iterator::create_charset_iterator(
     const Default_charset &default_charset,
-    const std::vector<uint> &column_charset)
+    const Dynamic_array<uint> &column_charset)
 {
   if (!default_charset.empty())
     return std::unique_ptr<Charset_iterator>(
@@ -3459,7 +3443,6 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
                                         const Optional_metadata_fields &fields)
 {
   unsigned char* field_metadata_ptr= m_field_metadata;
-  std::vector<bool>::const_iterator signedness_it= fields.m_signedness.begin();
 
   std::unique_ptr<Charset_iterator> charset_it =
       Charset_iterator::create_charset_iterator(fields.m_default_charset,
@@ -3468,13 +3451,11 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
       Charset_iterator::create_charset_iterator(
           fields.m_enum_and_set_default_charset,
           fields.m_enum_and_set_column_charset);
-  std::vector<Optional_metadata_fields::str_vector>::const_iterator
-    set_str_values_it= fields.m_set_str_value.begin();
-  std::vector<Optional_metadata_fields::str_vector>::const_iterator
-    enum_str_values_it= fields.m_enum_str_value.begin();
-  std::vector<unsigned int>::const_iterator geometry_type_it=
-    fields.m_geometry_type.begin();
-  LEX_CSTRING *col_names= fields.m_column_name;
+  const Optional_metadata_fields::str_vector *set_str_values_it=
+      fields.m_set_str_value.front();
+  const Optional_metadata_fields::str_vector *enum_str_values_it=
+      fields.m_enum_str_value.front();
+  const uint *geometry_type_it= fields.m_geometry_type.front();
 
   uint geometry_type= 0;
 
@@ -3482,6 +3463,7 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
 
   for (unsigned long i= 0; i < m_colcnt; i++)
   {
+    const Optional_metadata_fields::Column_metadata column_metadata = fields.m_column_metadata.at(i);
     uint real_type = m_coltype[i];
     if (real_type == MYSQL_TYPE_STRING &&
         (*field_metadata_ptr == MYSQL_TYPE_ENUM ||
@@ -3497,11 +3479,11 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
       cs = enum_and_set_charset_it->next();
 
     // Print column name
-    if (col_names && col_names->str)
+    const LEX_CSTRING& col_name= column_metadata.column_name;
+    if (col_name.str)
     {
-      pretty_print_identifier(file, col_names->str, col_names->length);
+      pretty_print_identifier(file, col_name.str, col_name.length);
       my_b_printf(file, " ");
-      col_names++;
     }
 
     // update geometry_type for geometry columns
@@ -3525,12 +3507,10 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
     my_b_printf(file, "%s", type_name);
 
     // Print UNSIGNED for numeric column
-    if (is_numeric_type(real_type) &&
-        signedness_it != fields.m_signedness.end())
+    if (is_numeric_type(real_type))
     {
-      if (*signedness_it == true)
+      if (column_metadata.is_unsigned == true)
         my_b_printf(file, " UNSIGNED");
-      signedness_it++;
     }
 
     // if the column is not marked as 'null', print 'not null'
@@ -3542,24 +3522,24 @@ void Table_map_log_event::print_columns(IO_CACHE *file,
     if (real_type == MYSQL_TYPE_ENUM &&
         enum_str_values_it != fields.m_enum_str_value.end())
     {
-      str_values= &(*enum_str_values_it);
+      str_values= enum_str_values_it;
       enum_str_values_it++;
     }
     else if (real_type == MYSQL_TYPE_SET &&
              set_str_values_it != fields.m_set_str_value.end())
     {
-      str_values= &(*set_str_values_it);
+      str_values= set_str_values_it;
       set_str_values_it++;
     }
 
     if (str_values != NULL)
     {
       const char *separator= "(";
-      for (Optional_metadata_fields::str_vector::const_iterator it=
-             str_values->begin(); it != str_values->end(); it++)
+      for (const LEX_CSTRING *it= str_values->front(); it != str_values->end();
+           it++)
       {
         my_b_printf(file, "%s", separator);
-        pretty_print_str(file, it->c_str(), it->size());
+        pretty_print_str(file, it->str, it->length);
         separator= ",";
       }
       my_b_printf(file, ")");
@@ -3582,19 +3562,18 @@ void Table_map_log_event::print_primary_key
   {
     my_b_printf(file, "# Primary Key(");
 
-    std::vector<Optional_metadata_fields::uint_pair>::const_iterator it=
-      fields.m_primary_key.begin();
-
-    for (; it != fields.m_primary_key.end(); it++)
+    for (auto *it= fields.m_primary_key.front();
+         it != fields.m_primary_key.end(); it++)
     {
-      if (it != fields.m_primary_key.begin())
+      if (it != fields.m_primary_key.front())
         my_b_printf(file, ", ");
 
       // Print column name or column index
-      if (!fields.m_column_name)
-        my_b_printf(file, "%u", it->first);
+      const LEX_CSTRING& column_name = fields.m_column_metadata.at(it->first).column_name;
+      if (column_name.str)
+        my_b_printf(file, "%s", column_name.str);
       else
-        my_b_printf(file, "%s", fields.m_column_name[it->first].str);
+        my_b_printf(file, "%u", it->first);
 
       // Print prefix length
       if (it->second != 0)
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index 10051d8d35e07..4e6f87d0c6e4c 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -94,6 +94,7 @@ class table_def
   {
     return static_cast<enum_field_types>(m_type[index]);
   }
+  const uchar *field_types() const { return m_type; }
   /*
     Return a representation of the type data for one field.
 
diff --git a/sql/rpl_utility_server.cc b/sql/rpl_utility_server.cc
index cade88c093024..17a5ece3f42a1 100644
--- a/sql/rpl_utility_server.cc
+++ b/sql/rpl_utility_server.cc
@@ -1351,10 +1351,12 @@ void RPL_TABLE_LIST::create_column_mapping(rpl_group_info *rgi)
 
   Table_map_log_event::Optional_metadata_fields
     opt_metadata(rgi->thd->mem_root, master_cols,
+                 m_tabledef.field_types(),
                  (uchar*) m_tabledef.optional_metadata.str,
                  m_tabledef.optional_metadata.length, 1);
 
-  if (!opt_metadata.m_column_name)
+  if (!opt_metadata.m_column_metadata.empty() && 
+          !opt_metadata.m_column_metadata.at(0).column_name.str)
   {
     /*
       If there are no column names provided in the optional metadata
@@ -1366,7 +1368,7 @@ void RPL_TABLE_LIST::create_column_mapping(rpl_group_info *rgi)
 
   for (uint col= 0; col < master_cols; col++)
   {
-    const LEX_CSTRING *field_name= &opt_metadata.m_column_name[col];
+    const LEX_CSTRING *field_name= &opt_metadata.m_column_metadata.at(col).column_name;
     Field *field= table->find_field_by_name(field_name);
     if (unlikely(!field))
     {
diff --git a/sql/sql_array.h b/sql/sql_array.h
index 812e8aae12b6d..8e72244a98f8c 100644
--- a/sql/sql_array.h
+++ b/sql/sql_array.h
@@ -42,7 +42,7 @@ template <typename Element_type> class Bounds_checked_array
   {}
 
   void reset() { m_array= NULL; m_size= 0; }
- 
+
   void reset(Element_type *array_arg, size_t size_arg)
   {
     m_array= array_arg;
@@ -197,6 +197,8 @@ template <class Elem> class Dynamic_array
 
   size_t size() const { return array.elements; }
 
+  bool empty() const { return array.elements == 0; }
+
   const Elem *end() const
   {
     return back() + 1;
@@ -282,7 +284,7 @@ template <class Elem> class Dynamic_array
     size_t old_size= elements();
     if (reserve(new_size))
       return true;
-    
+
     if (new_size > old_size)
     {
       set_dynamic(&array, (uchar*)&default_val, new_size - 1);