MDEV-18408 Assertion `0' failed in Item::val_native_result / Timestamp_or_zero_datetime_native_null::Timestamp_or_zero_datetime_native_null upon mysqld_list_fields after crash recovery

The problem happened because Item_ident_for_show did not implement val_native().

Solution:

- Removing class Item_ident_for_show
- Implementing a new method Protocol::send_list_fields() instead,
  which accepts a List<Field> instead of List<Item> as input.
  Now no any Item creation is done during mysqld_list_fields().

Adding helper methods, to reuse the code easier:
- Moved a part of Protocol::send_result_set_metadata(),
  responsible for sending an individual field metadata,
  into a new method Protocol_text::store_field_metadata().
  Reusing it in both send_list_fields() and send_result_set_metadata().
- Adding Protocol_text::store_field_metadata()
- Adding Protocol_text::store_field_metadata_for_list_fields()

Note, this patch also automatically fixed another bug:

MDEV-18685 mysql_list_fields() returns DEFAULT 0 instead of DEFAULT NULL for view columns

The reason for this bug was that Item_ident_for_show::val_xxx() and get_date()
did not check field->is_null() before calling field->val_xxx()/get_date().
Now the default value is correctly sent by Protocol_text::store(Field*).

8 files changed, 411 insertions, 271 deletions

diff --git a/libmysqld/lib_sql.cc b/libmysqld/lib_sql.cc
index 305f6346c9e..30e40246cb0 100644
--- a/libmysqld/lib_sql.cc
+++ b/libmysqld/lib_sql.cc
@@ -891,6 +891,20 @@ static char *dup_str_aux(MEM_ROOT *root, const char *from, uint length,
 }
 
 
+static char *dup_str_aux(MEM_ROOT *root, const char *from,
+                         CHARSET_INFO *fromcs, CHARSET_INFO *tocs)
+{
+  return dup_str_aux(root, from, (uint) strlen(from), fromcs, tocs);
+}
+
+
+static char *dup_str_aux(MEM_ROOT *root, const LEX_CSTRING &from,
+                         CHARSET_INFO *fromcs, CHARSET_INFO *tocs)
+{
+  return dup_str_aux(root, from.str, (uint) from.length, fromcs, tocs);
+}
+
+
 /*
   creates new result and hooks it to the list
 
@@ -969,7 +983,7 @@ write_eof_packet(THD *thd, uint server_status, uint statement_warn_count)
     1 if memory allocation failed
 */
 
-int Protocol::begin_dataset()
+bool Protocol::begin_dataset()
 {
   MYSQL_DATA *data= thd->alloc_new_dataset();
   if (!data)
@@ -982,6 +996,19 @@ int Protocol::begin_dataset()
 }
 
 
+bool Protocol::begin_dataset(THD *thd, uint numfields)
+{
+  if (begin_dataset())
+    return true;
+  MYSQL_DATA *data= thd->cur_data;
+  data->fields= field_count= numfields;
+  if (!(data->embedded_info->fields_list=
+      (MYSQL_FIELD*)alloc_root(&data->alloc, sizeof(MYSQL_FIELD)*field_count)))
+    return true;
+  return false;
+}
+
+
 /*
   remove last row of current recordset
 
@@ -1011,110 +1038,80 @@ void Protocol_text::remove_last_row()
 }
 
 
+bool Protocol_text::store_field_metadata(const THD * thd,
+                                         const Send_field &server_field,
+                                         CHARSET_INFO *charset_for_protocol,
+                                         uint pos)
+{
+  CHARSET_INFO *cs= system_charset_info;
+  CHARSET_INFO *thd_cs= thd->variables.character_set_results;
+  MYSQL_DATA *data= thd->cur_data;
+  MEM_ROOT *field_alloc= &data->alloc;
+  MYSQL_FIELD *client_field= &thd->cur_data->embedded_info->fields_list[pos];
+  DBUG_ASSERT(server_field.is_sane());
+
+  client_field->db= dup_str_aux(field_alloc, server_field.db_name,
+                                cs, thd_cs);
+  client_field->table= dup_str_aux(field_alloc, server_field.table_name,
+                                   cs, thd_cs);
+  client_field->name= dup_str_aux(field_alloc, server_field.col_name,
+                                  cs, thd_cs);
+  client_field->org_table= dup_str_aux(field_alloc, server_field.org_table_name,
+                                       cs, thd_cs);
+  client_field->org_name= dup_str_aux(field_alloc, server_field.org_col_name,
+                                      cs, thd_cs);
+  if (charset_for_protocol == &my_charset_bin || thd_cs == NULL)
+  {
+    /* No conversion */
+    client_field->charsetnr= charset_for_protocol->number;
+    client_field->length= server_field.length;
+  }
+  else
+  {
+    /* With conversion */
+    client_field->charsetnr= thd_cs->number;
+    client_field->length= server_field.max_octet_length(charset_for_protocol,
+                                                        thd_cs);
+  }
+  client_field->type= server_field.type;
+  client_field->flags= (uint16) server_field.flags;
+  client_field->decimals= server_field.decimals;
+
+  client_field->db_length=		strlen(client_field->db);
+  client_field->table_length=		strlen(client_field->table);
+  client_field->name_length=		strlen(client_field->name);
+  client_field->org_name_length=	strlen(client_field->org_name);
+  client_field->org_table_length=	strlen(client_field->org_table);
+
+  client_field->catalog= dup_str_aux(field_alloc, "def", 3, cs, thd_cs);
+  client_field->catalog_length= 3;
+
+  if (IS_NUM(client_field->type))
+    client_field->flags|= NUM_FLAG;
+
+  client_field->max_length= 0;
+  client_field->def= 0;
+  return false;
+}
+
+
 bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
 {
   List_iterator_fast<Item> it(*list);
   Item                     *item;
-  MYSQL_FIELD              *client_field;
-  MEM_ROOT                 *field_alloc;
-  CHARSET_INFO             *thd_cs= thd->variables.character_set_results;
-  CHARSET_INFO             *cs= system_charset_info;
-  MYSQL_DATA               *data;
+  Protocol_text prot(thd);
   DBUG_ENTER("send_result_set_metadata");
 
   if (!thd->mysql)            // bootstrap file handling
     DBUG_RETURN(0);
 
-  if (begin_dataset())
-    goto err;
-
-  data= thd->cur_data;
-  data->fields= field_count= list->elements;
-  field_alloc= &data->alloc;
-
-  if (!(client_field= data->embedded_info->fields_list= 
-	(MYSQL_FIELD*)alloc_root(field_alloc, sizeof(MYSQL_FIELD)*field_count)))
+  if (begin_dataset(thd, list->elements))
     goto err;
 
-  while ((item= it++))
+  for (uint pos= 0 ; (item= it++); pos++)
   {
-    Send_field server_field;
-    item->make_send_field(thd, &server_field);
-
-    /* Keep things compatible for old clients */
-    if (server_field.type == MYSQL_TYPE_VARCHAR)
-      server_field.type= MYSQL_TYPE_VAR_STRING;
-
-    client_field->db= dup_str_aux(field_alloc, server_field.db_name,
-                                  strlen(server_field.db_name), cs, thd_cs);
-    client_field->table= dup_str_aux(field_alloc, server_field.table_name,
-                                     strlen(server_field.table_name), cs, thd_cs);
-    client_field->name= dup_str_aux(field_alloc, server_field.col_name.str,
-                                    server_field.col_name.length, cs, thd_cs);
-    client_field->org_table= dup_str_aux(field_alloc, server_field.org_table_name,
-                                         strlen(server_field.org_table_name), cs, thd_cs);
-    client_field->org_name= dup_str_aux(field_alloc,
-                                        server_field.org_col_name.str,
-                                        server_field.org_col_name.length,
-                                        cs, thd_cs);
-    if (item->charset_for_protocol() == &my_charset_bin || thd_cs == NULL)
-    {
-      /* No conversion */
-      client_field->charsetnr= item->charset_for_protocol()->number;
-      client_field->length= server_field.length;
-    }
-    else
-    {
-      uint max_char_len;
-      /* With conversion */
-      client_field->charsetnr= thd_cs->number;
-      max_char_len= (server_field.type >= (int) MYSQL_TYPE_TINY_BLOB &&
-                     server_field.type <= (int) MYSQL_TYPE_BLOB) ?
-                     server_field.length / item->collation.collation->mbminlen :
-                     server_field.length / item->collation.collation->mbmaxlen;
-      client_field->length= char_to_byte_length_safe(max_char_len,
-                                                     thd_cs->mbmaxlen);
-    }
-    client_field->type=   server_field.type;
-    client_field->flags= (uint16) server_field.flags;
-    client_field->decimals= server_field.decimals;
-    if (server_field.type == MYSQL_TYPE_FLOAT ||
-        server_field.type == MYSQL_TYPE_DOUBLE)
-      set_if_smaller(client_field->decimals, FLOATING_POINT_DECIMALS);
-
-    client_field->db_length=		strlen(client_field->db);
-    client_field->table_length=		strlen(client_field->table);
-    client_field->name_length=		strlen(client_field->name);
-    client_field->org_name_length=	strlen(client_field->org_name);
-    client_field->org_table_length=	strlen(client_field->org_table);
-
-    client_field->catalog= dup_str_aux(field_alloc, "def", 3, cs, thd_cs);
-    client_field->catalog_length= 3;
-
-    if (IS_NUM(client_field->type))
-      client_field->flags|= NUM_FLAG;
-
-    if (flags & (int) Protocol::SEND_DEFAULTS)
-    {
-      char buff[80];
-      String tmp(buff, sizeof(buff), default_charset_info), *res;
-
-      if (!(res=item->val_str(&tmp)))
-      {
-	client_field->def_length= 0;
-	client_field->def= strmake_root(field_alloc, "",0);
-      }
-      else
-      {
-	client_field->def_length= res->length();
-	client_field->def= strmake_root(field_alloc, res->ptr(),
-					client_field->def_length);
-      }
-    }
-    else
-      client_field->def=0;
-    client_field->max_length= 0;
-    ++client_field;
+    if (prot.store_field_metadata(thd, item, pos))
+      goto err;
   }
 
   if (flags & SEND_EOF)
@@ -1127,6 +1124,55 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
   DBUG_RETURN(1);				/* purecov: inspected */
 }
 
+
+static void
+list_fields_send_default(THD *thd, Protocol *p, Field *fld, uint pos)
+{
+  char buff[80];
+  String tmp(buff, sizeof(buff), default_charset_info), *res;
+  MYSQL_FIELD *client_field= &thd->cur_data->embedded_info->fields_list[pos];
+
+  if (fld->is_null() || !(res= fld->val_str(&tmp)))
+  {
+    client_field->def_length= 0;
+    client_field->def= strmake_root(&thd->cur_data->alloc, "", 0);
+  }
+  else
+  {
+    client_field->def_length= res->length();
+    client_field->def= strmake_root(&thd->cur_data->alloc, res->ptr(),
+                                    client_field->def_length);
+  }
+}
+
+
+bool Protocol::send_list_fields(List<Field> *list, const TABLE_LIST *table_list)
+{
+  DBUG_ENTER("send_result_set_metadata");
+  Protocol_text prot(thd);
+  List_iterator_fast<Field> it(*list);
+  Field *fld;
+
+  if (!thd->mysql)            // bootstrap file handling
+    DBUG_RETURN(0);
+
+  if (begin_dataset(thd, list->elements))
+    goto err;
+
+  for (uint pos= 0 ; (fld= it++); pos++)
+  {
+    if (prot.store_field_metadata_for_list_fields(thd, fld, table_list, pos))
+      goto err;
+    list_fields_send_default(thd, this, fld, pos);
+  }
+
+  DBUG_RETURN(prepare_for_send(list->elements));
+err:
+  my_error(ER_OUT_OF_RESOURCES, MYF(0));
+  DBUG_RETURN(1);
+}
+
+
 bool Protocol::write()
 {
   if (!thd->mysql)            // bootstrap file handling
diff --git a/sql/field.h b/sql/field.h
index 97926a356d1..aad96adcd40 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -4958,6 +4958,83 @@ class Send_field :public Sql_alloc {
   uint flags, decimals;
   enum_field_types type;
   Send_field() {}
+  Send_field(Field *field)
+  {
+    field->make_send_field(this);
+    DBUG_ASSERT(table_name != 0);
+    normalize();
+  }
+
+  Send_field(Field *field,
+             const char *db_name_arg,
+             const char *table_name_arg)
+   :db_name(db_name_arg),
+    table_name(table_name_arg),
+    org_table_name(table_name_arg),
+    col_name(field->field_name),
+    org_col_name(field->field_name),
+    length(field->field_length),
+    flags(field->table->maybe_null ?
+          (field->flags & ~NOT_NULL_FLAG) : field->flags),
+    decimals(field->decimals()),
+    type(field->type())
+  {
+    normalize();
+  }
+
+  // This should move to Type_handler eventually
+  static enum_field_types protocol_type_code(enum_field_types type)
+  {
+    /* Keep things compatible for old clients */
+    if (type == MYSQL_TYPE_VARCHAR)
+      return MYSQL_TYPE_VAR_STRING;
+    return type;
+  }
+  void normalize()
+  {
+    /* limit number of decimals for float and double */
+    if (type == MYSQL_TYPE_FLOAT || type == MYSQL_TYPE_DOUBLE)
+      set_if_smaller(decimals, FLOATING_POINT_DECIMALS);
+    /* Keep things compatible for old clients */
+    type= protocol_type_code(type);
+  }
+
+  // This should move to Type_handler eventually
+  uint32 max_char_length(CHARSET_INFO *cs) const
+  {
+    return type >= MYSQL_TYPE_TINY_BLOB && type <= MYSQL_TYPE_BLOB ?
+                   length / cs->mbminlen :
+                   length / cs->mbmaxlen;
+  }
+  uint32 max_octet_length(CHARSET_INFO *from, CHARSET_INFO *to) const
+  {
+    /*
+      For TEXT/BLOB columns, field_length describes the maximum data
+      length in bytes. There is no limit to the number of characters
+      that a TEXT column can store, as long as the data fits into
+      the designated space.
+      For the rest of textual columns, field_length is evaluated as
+      char_count * mbmaxlen, where character count is taken from the
+      definition of the column. In other words, the maximum number
+      of characters here is limited by the column definition.
+
+      When one has a LONG TEXT column with a single-byte
+      character set, and the connection character set is multi-byte, the
+      client may get fields longer than UINT_MAX32, due to
+      <character set column> -> <character set connection> conversion.
+      In that case column max length would not fit into the 4 bytes
+      reserved for it in the protocol. So we cut it here to UINT_MAX32.
+    */
+    return char_to_byte_length_safe(max_char_length(from), to->mbmaxlen);
+  }
+
+  // This should move to Type_handler eventually
+  bool is_sane() const
+  {
+    return (decimals <= FLOATING_POINT_DECIMALS ||
+            (type != MYSQL_TYPE_FLOAT && type != MYSQL_TYPE_DOUBLE)) &&
+           type != MYSQL_TYPE_VARCHAR;
+  }
 };
 
 
diff --git a/sql/item.cc b/sql/item.cc
index 47e10ba4004..6589f1e5683 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -2831,18 +2831,6 @@ Item* Item_ref::build_clone(THD *thd)
 }
 
 
-void Item_ident_for_show::make_send_field(THD *thd, Send_field *tmp_field)
-{
-  tmp_field->table_name= tmp_field->org_table_name= table_name;
-  tmp_field->db_name= db_name;
-  tmp_field->col_name= tmp_field->org_col_name= field->field_name;
-  tmp_field->length=field->field_length;
-  tmp_field->type=field->type();
-  tmp_field->flags= field->table->maybe_null ? 
-    (field->flags & ~NOT_NULL_FLAG) : field->flags;
-  tmp_field->decimals= field->decimals();
-}
-
 /**********************************************/
 
 Item_field::Item_field(THD *thd, Field *f)
diff --git a/sql/item.h b/sql/item.h
index 2691a69a14b..0ffbd2f8bc4 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -1298,7 +1298,6 @@ public:
    /*
      The default implementation for the Items that do not need native format:
      - Item_basic_value
-     - Item_ident_for_show
      - Item_copy
      - Item_exists_subselect
      - Item_sum_field
@@ -3239,45 +3238,6 @@ public:
 };
 
 
-class Item_ident_for_show :public Item
-{
-public:
-  Field *field;
-  const char *db_name;
-  const char *table_name;
-
-  Item_ident_for_show(THD *thd, Field *par_field, const char *db_arg,
-                      const char *table_name_arg):
-    Item(thd), field(par_field), db_name(db_arg), table_name(table_name_arg)
-  {
-    Type_std_attributes::set(par_field->type_std_attributes());
-  }
-  enum Type type() const { return FIELD_ITEM; }
-  Field *create_tmp_field_ex(TABLE *table, Tmp_field_src *src,
-                             const Tmp_field_param *param)
-  {
-    DBUG_ASSERT(0);
-    return 0;
-  }
-  double val_real() { return field->val_real(); }
-  longlong val_int() { return field->val_int(); }
-  String *val_str(String *str) { return field->val_str(str); }
-  my_decimal *val_decimal(my_decimal *dec) { return field->val_decimal(dec); }
-  bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
-  {
-    return field->get_date(ltime, fuzzydate);
-  }
-  void make_send_field(THD *thd, Send_field *tmp_field);
-  const Type_handler *type_handler() const
-  {
-    const Type_handler *handler= field->type_handler();
-    return handler->type_handler_for_item_field();
-  }
-  Item* get_copy(THD *thd)
-  { return get_item_copy<Item_ident_for_show>(thd, this); }
-};
-
-
 class Item_field :public Item_ident,
                   public Load_data_outvar
 {
diff --git a/sql/protocol.cc b/sql/protocol.cc
index 38eb8ac99f7..1d5f77fb995 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -788,6 +788,73 @@ bool Protocol::flush()
 
 #ifndef EMBEDDED_LIBRARY
 
+bool Protocol_text::store_field_metadata(const THD * thd,
+                                         const Send_field &field,
+                                         CHARSET_INFO *charset_for_protocol,
+                                         uint fieldnr)
+{
+  CHARSET_INFO *thd_charset= thd->variables.character_set_results;
+  char *pos;
+  CHARSET_INFO *cs= system_charset_info;
+  DBUG_ASSERT(field.is_sane());
+
+  if (thd->client_capabilities & CLIENT_PROTOCOL_41)
+  {
+    if (store(STRING_WITH_LEN("def"), cs, thd_charset) ||
+        store_str(field.db_name, cs, thd_charset) ||
+        store_str(field.table_name, cs, thd_charset) ||
+        store_str(field.org_table_name, cs, thd_charset) ||
+        store_str(field.col_name, cs, thd_charset) ||
+        store_str(field.org_col_name, cs, thd_charset) ||
+        packet->realloc(packet->length() + 12))
+      return true;
+    /* Store fixed length fields */
+    pos= (char*) packet->end();
+    *pos++= 12;                                // Length of packed fields
+    /* inject a NULL to test the client */
+    DBUG_EXECUTE_IF("poison_rs_fields", pos[-1]= (char) 0xfb;);
+    if (charset_for_protocol == &my_charset_bin || thd_charset == NULL)
+    {
+      /* No conversion */
+      int2store(pos, charset_for_protocol->number);
+      int4store(pos + 2, field.length);
+    }
+    else
+    {
+      /* With conversion */
+      int2store(pos, thd_charset->number);
+      uint32 field_length= field.max_octet_length(charset_for_protocol,
+                                                  thd_charset);
+      int4store(pos + 2, field_length);
+    }
+    pos[6]= field.type;
+    int2store(pos + 7, field.flags);
+    pos[9]= (char) field.decimals;
+    pos[10]= 0;                                // For the future
+    pos[11]= 0;                                // For the future
+    pos+= 12;
+  }
+  else
+  {
+    if (store_str(field.table_name, cs, thd_charset) ||
+        store_str(field.col_name, cs, thd_charset) ||
+        packet->realloc(packet->length() + 10))
+      return true;
+    pos= (char*) packet->end();
+    pos[0]= 3;
+    int3store(pos + 1, field.length);
+    pos[4]= 1;
+    pos[5]= field.type;
+    pos[6]= 3;
+    int2store(pos + 7, field.flags);
+    pos[9]= (char) field.decimals;
+    pos+= 10;
+  }
+  packet->length((uint) (pos - packet->ptr()));
+  return false;
+}
+
+
 /**
   Send name and type of result to client.
 
@@ -810,10 +877,7 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
 {
   List_iterator_fast<Item> it(*list);
   Item *item;
-  ValueBuffer<MAX_FIELD_WIDTH> tmp;
-  Protocol_text prot(thd);
-  String *local_packet= prot.storage_packet();
-  CHARSET_INFO *thd_charset= thd->variables.character_set_results;
+  Protocol_text prot(thd, thd->variables.net_buffer_length);
   DBUG_ENTER("Protocol::send_result_set_metadata");
 
   if (flags & SEND_NUM_ROWS)
@@ -828,117 +892,17 @@ bool Protocol::send_result_set_metadata(List<Item> *list, uint flags)
 #ifndef DBUG_OFF
   field_types= (enum_field_types*) thd->alloc(sizeof(field_types) *
 					      list->elements);
-  uint count= 0;
 #endif
 
-  /* We have to reallocate it here as a stored procedure may have reset it */
-  (void) local_packet->alloc(thd->variables.net_buffer_length);
-
-  while ((item=it++))
+  for (uint pos= 0; (item=it++); pos++)
   {
-    char *pos;
-    CHARSET_INFO *cs= system_charset_info;
-    Send_field field;
-    item->make_send_field(thd, &field);
-
-    /* limit number of decimals for float and double */
-    if (field.type == MYSQL_TYPE_FLOAT || field.type == MYSQL_TYPE_DOUBLE)
-      set_if_smaller(field.decimals, FLOATING_POINT_DECIMALS);
-
-    /* Keep things compatible for old clients */
-    if (field.type == MYSQL_TYPE_VARCHAR)
-      field.type= MYSQL_TYPE_VAR_STRING;
-
     prot.prepare_for_resend();
-
-    if (thd->client_capabilities & CLIENT_PROTOCOL_41)
-    {
-      if (prot.store(STRING_WITH_LEN("def"), cs, thd_charset) ||
-	  prot.store(field.db_name, (uint) strlen(field.db_name),
-		     cs, thd_charset) ||
-	  prot.store(field.table_name, (uint) strlen(field.table_name),
-		     cs, thd_charset) ||
-	  prot.store(field.org_table_name, (uint) strlen(field.org_table_name),
-		     cs, thd_charset) ||
-	  prot.store(field.col_name.str, (uint) field.col_name.length,
-		     cs, thd_charset) ||
-	  prot.store(field.org_col_name.str, (uint) field.org_col_name.length,
-		     cs, thd_charset) ||
-	  local_packet->realloc(local_packet->length()+12))
-	goto err;
-      /* Store fixed length fields */
-      pos= (char*) local_packet->ptr()+local_packet->length();
-      *pos++= 12;				// Length of packed fields
-      /* inject a NULL to test the client */
-      DBUG_EXECUTE_IF("poison_rs_fields", pos[-1]= (char) 0xfb;);
-      if (item->charset_for_protocol() == &my_charset_bin || thd_charset == NULL)
-      {
-        /* No conversion */
-        int2store(pos, item->charset_for_protocol()->number);
-        int4store(pos+2, field.length);
-      }
-      else
-      {
-        /* With conversion */
-        uint32 field_length, max_length;
-        int2store(pos, thd_charset->number);
-        /*
-          For TEXT/BLOB columns, field_length describes the maximum data
-          length in bytes. There is no limit to the number of characters
-          that a TEXT column can store, as long as the data fits into
-          the designated space.
-          For the rest of textual columns, field_length is evaluated as
-          char_count * mbmaxlen, where character count is taken from the
-          definition of the column. In other words, the maximum number
-          of characters here is limited by the column definition.
-
-          When one has a LONG TEXT column with a single-byte
-          character set, and the connection character set is multi-byte, the
-          client may get fields longer than UINT_MAX32, due to
-          <character set column> -> <character set connection> conversion.
-          In that case column max length does not fit into the 4 bytes
-          reserved for it in the protocol.
-        */
-        max_length= (field.type >= MYSQL_TYPE_TINY_BLOB &&
-                     field.type <= MYSQL_TYPE_BLOB) ?
-                     field.length / item->collation.collation->mbminlen :
-                     field.length / item->collation.collation->mbmaxlen;
-        field_length= char_to_byte_length_safe(max_length,
-                                               thd_charset->mbmaxlen);
-        int4store(pos + 2, field_length);
-      }
-      pos[6]= field.type;
-      int2store(pos+7,field.flags);
-      pos[9]= (char) field.decimals;
-      pos[10]= 0;				// For the future
-      pos[11]= 0;				// For the future
-      pos+= 12;
-    }
-    else
-    {
-      if (prot.store(field.table_name, (uint) strlen(field.table_name),
-		     cs, thd_charset) ||
-	  prot.store(field.col_name.str, (uint) field.col_name.length,
-		     cs, thd_charset) ||
-	  local_packet->realloc(local_packet->length()+10))
-	goto err;
-      pos= (char*) local_packet->ptr()+local_packet->length();
-      pos[0]=3;
-      int3store(pos+1,field.length);
-      pos[4]=1;
-      pos[5]=field.type;
-      pos[6]=3;
-      int2store(pos+7,field.flags);
-      pos[9]= (char) field.decimals;
-      pos+= 10;
-    }
-    local_packet->length((uint) (pos - local_packet->ptr()));
-    if (flags & SEND_DEFAULTS)
-      item->send(&prot, &tmp);			// Send default value
+    if (prot.store_field_metadata(thd, item, pos))
+      goto err;
     if (prot.write())
       DBUG_RETURN(1);
 #ifndef DBUG_OFF
-    field_types[count++]= field.type;
+    field_types[pos]= Send_field::protocol_type_code(item->field_type());
 #endif
   }
 
@@ -967,6 +931,38 @@ err:
 }
 
 
+bool Protocol::send_list_fields(List<Field> *list, const TABLE_LIST *table_list)
+{
+  DBUG_ENTER("Protocol::send_list_fields");
+  List_iterator_fast<Field> it(*list);
+  Field *fld;
+  Protocol_text prot(thd, thd->variables.net_buffer_length);
+
+#ifndef DBUG_OFF
+  field_types= (enum_field_types*) thd->alloc(sizeof(field_types) *
+                                              list->elements);
+#endif
+
+  for (uint pos= 0; (fld= it++); pos++)
+  {
+    prot.prepare_for_resend();
+    if (prot.store_field_metadata_for_list_fields(thd, fld, table_list, pos))
+      goto err;
+    prot.store(fld);   // Send default value
+    if (prot.write())
+      DBUG_RETURN(1);
+#ifndef DBUG_OFF
+    field_types[pos]= Send_field::protocol_type_code(fld->type());
+#endif
+  }
+  DBUG_RETURN(prepare_for_send(list->elements));
+
+err:
+  my_message(ER_OUT_OF_RESOURCES, ER_THD(thd, ER_OUT_OF_RESOURCES), MYF(0));
+  DBUG_RETURN(1);
+}
+
+
 bool Protocol::write()
 {
   DBUG_ENTER("Protocol::write");
@@ -976,6 +972,27 @@ bool Protocol::write()
 #endif /* EMBEDDED_LIBRARY */
 
 
+bool Protocol_text::store_field_metadata(THD *thd, Item *item, uint pos)
+{
+  Send_field field;
+  item->make_send_field(thd, &field);
+  field.normalize();
+  return store_field_metadata(thd, field, item->charset_for_protocol(), pos);
+}
+
+
+bool Protocol_text::store_field_metadata_for_list_fields(const THD *thd,
+                                                         Field *fld,
+                                                         const TABLE_LIST *tl,
+                                                         uint pos)
+{
+  Send_field field= tl->view ?
+                    Send_field(fld, tl->view_db.str, tl->view_name.str) :
+                    Send_field(fld);
+  return store_field_metadata(thd, field, fld->charset_for_protocol(), pos);
+}
+
+
 /**
   Send one result set row.
 
diff --git a/sql/protocol.h b/sql/protocol.h
index 6397e3dd5e6..6dd3e5c7521 100644
--- a/sql/protocol.h
+++ b/sql/protocol.h
@@ -25,8 +25,10 @@
 
 class i_string;
 class Field;
+class Send_field;
 class THD;
 class Item_param;
+struct TABLE_LIST;
 typedef struct st_mysql_field MYSQL_FIELD;
 typedef struct st_mysql_rows MYSQL_ROWS;
 
@@ -75,8 +77,9 @@ public:
   virtual ~Protocol() {}
   void init(THD* thd_arg);
 
-  enum { SEND_NUM_ROWS= 1, SEND_DEFAULTS= 2, SEND_EOF= 4 };
+  enum { SEND_NUM_ROWS= 1, SEND_EOF= 2 };
   virtual bool send_result_set_metadata(List<Item> *list, uint flags);
+  bool send_list_fields(List<Field> *list, const TABLE_LIST *table_list);
   bool send_result_set_row(List<Item> *row_items);
 
   bool store(I_List<i_string> *str_list);
@@ -113,6 +116,15 @@ public:
   virtual bool store(const char *from, size_t length, CHARSET_INFO *cs)=0;
   virtual bool store(const char *from, size_t length, 
   		     CHARSET_INFO *fromcs, CHARSET_INFO *tocs)=0;
+  bool store_str(const char *s, CHARSET_INFO *fromcs, CHARSET_INFO *tocs)
+  {
+    DBUG_ASSERT(s);
+    return store(s, (uint) strlen(s), fromcs, tocs);
+  }
+  bool store_str(const LEX_CSTRING &s, CHARSET_INFO *fromcs, CHARSET_INFO *tocs)
+  {
+    return store(s.str, (uint) s.length, fromcs, tocs);
+  }
   virtual bool store(float from, uint32 decimals, String *buffer)=0;
   virtual bool store(double from, uint32 decimals, String *buffer)=0;
   virtual bool store(MYSQL_TIME *time, int decimals)=0;
@@ -122,7 +134,8 @@ public:
 
   virtual bool send_out_parameters(List<Item_param> *sp_params)=0;
 #ifdef EMBEDDED_LIBRARY
-  int begin_dataset();
+  bool begin_dataset();
+  bool begin_dataset(THD *thd, uint numfields);
   virtual void remove_last_row() {}
 #else
   void remove_last_row() {}
@@ -150,7 +163,12 @@ public:
 class Protocol_text :public Protocol
 {
 public:
-  Protocol_text(THD *thd_arg) :Protocol(thd_arg) {}
+  Protocol_text(THD *thd_arg, ulong prealloc= 0)
+   :Protocol(thd_arg)
+  {
+    if (prealloc)
+      packet->alloc(prealloc);
+  }
   virtual void prepare_for_resend();
   virtual bool store_null();
   virtual bool store_tiny(longlong from);
@@ -172,6 +190,13 @@ public:
 #ifdef EMBEDDED_LIBRARY
   void remove_last_row();
 #endif
+  bool store_field_metadata(const THD *thd, const Send_field &field,
+                            CHARSET_INFO *charset_for_protocol,
+                            uint pos);
+  bool store_field_metadata(THD *thd, Item *item, uint pos);
+  bool store_field_metadata_for_list_fields(const THD *thd, Field *field,
+                                            const TABLE_LIST *table_list,
+                                            uint pos);
   virtual enum enum_protocol_type type() { return PROTOCOL_TEXT; };
 };
 
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index c7b4fc6b2a7..07ef2b675f2 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1521,7 +1521,6 @@ void
 mysqld_list_fields(THD *thd, TABLE_LIST *table_list, const char *wild)
 {
   TABLE *table;
-  MEM_ROOT *mem_root= thd->mem_root;
   DBUG_ENTER("mysqld_list_fields");
   DBUG_PRINT("enter",("table: %s", table_list->table_name.str));
 
@@ -1531,28 +1530,18 @@ mysqld_list_fields(THD *thd, TABLE_LIST *table_list, const char *wild)
     DBUG_VOID_RETURN;
   table= table_list->table;
 
-  List<Item> field_list;
+  List<Field> field_list;
 
   Field **ptr,*field;
   for (ptr=table->field ; (field= *ptr); ptr++)
   {
     if (!wild || !wild[0] ||
         !wild_case_compare(system_charset_info, field->field_name.str,wild))
-    {
-      if (table_list->view)
-        field_list.push_back(new (mem_root)
-                             Item_ident_for_show(thd, field,
-                                                 table_list->view_db.str,
-                                                 table_list->view_name.str),
-                             mem_root);
-      else
-        field_list.push_back(new (mem_root) Item_field(thd, field), mem_root);
-    }
+      field_list.push_back(field);
   }
   restore_record(table, s->default_values);              // Get empty record
   table->use_all_columns();
-  if (thd->protocol->send_result_set_metadata(&field_list,
-                                              Protocol::SEND_DEFAULTS))
+  if (thd->protocol->send_list_fields(&field_list, table_list))
     DBUG_VOID_RETURN;
   my_eof(thd);
   DBUG_VOID_RETURN;
diff --git a/tests/mysql_client_test.c b/tests/mysql_client_test.c
index 2d238d84d54..6d17fa74005 100644
--- a/tests/mysql_client_test.c
+++ b/tests/mysql_client_test.c
@@ -8467,6 +8467,43 @@ static void test_list_fields_default()
 }
 
 
+/**
+  Note, this test covers MDEV-18408 and MDEV-18685
+*/
+
+static void test_mdev18408()
+{
+  MYSQL_RES *result;
+  int rc;
+  myheader("test_mdev18408s");
+
+  rc= mysql_query(mysql, "DROP TABLE IF EXISTS t1");
+  myquery(rc);
+
+  rc= mysql_query(mysql, "DROP VIEW IF EXISTS v1");
+  myquery(rc);
+
+  rc= mysql_query(mysql, "CREATE TABLE t1 (c1 TIMESTAMP NULL DEFAULT NULL)");
+  myquery(rc);
+
+  rc= mysql_query(mysql, "CREATE VIEW v1 AS SELECT c1 FROM t1");
+  myquery(rc);
+
+  result= mysql_list_fields(mysql, "v1", NULL);
+  mytest(result);
+
+  rc= my_process_result_set(result);
+  DIE_UNLESS(rc == 0);
+
+  verify_prepare_field(result, 0, "c1", "c1", MYSQL_TYPE_TIMESTAMP,
+                       "v1", "v1",  current_db, 19, 0);
+
+  mysql_free_result(result);
+  myquery(mysql_query(mysql, "DROP VIEW v1"));
+  myquery(mysql_query(mysql, "DROP TABLE t1"));
+}
+
+
 static void test_bug19671()
 {
   MYSQL_RES *result;
@@ -8493,7 +8530,7 @@ static void test_bug19671()
   DIE_UNLESS(rc == 0);
 
   verify_prepare_field(result, 0, "f1", "f1", MYSQL_TYPE_LONG,
-                       "v1", "v1", current_db, 11, "0");
+                       "v1", "v1", current_db, 11, NULL);
 
   mysql_free_result(result);
   myquery(mysql_query(mysql, "drop view v1"));
@@ -20977,6 +21014,7 @@ static struct my_tests_st my_tests[]= {
   { "test_bulk_delete", test_bulk_delete },
 #endif
   { "test_explain_meta", test_explain_meta },
+  { "test_mdev18408", test_mdev18408 },
   { 0, 0 }
 };