MDEV-18408 Assertion `0' failed in Item::val_native_result / Timestam…

…p_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*).

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

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;
+  }
 };
 
 

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)