MDEV-20042 Implement EXTRA2_FIELD_DATA_TYPE_INFO in FRM

mysql-test/main/frm-debug.result

+#
+# MDEV-20042 Implement EXTRA2_FIELD_DATA_TYPE_INFO in FRM
+#
+SET SESSION debug_dbug="+d,frm_data_type_info";
+CREATE TABLE t1 (c01 INT, c02 CHAR(20), c03 TEXT, c04 DOUBLE);
+Warnings:
+Note	1105	build_frm_image: Field data type info length: 0
+DROP TABLE t1;
+SET SESSION debug_dbug="-d,frm_data_type_info";
+SET SESSION debug_dbug="+d,frm_data_type_info";
+SET SESSION debug_dbug="+d,frm_data_type_info_emulate";
+CREATE TABLE t1 (c01 INT, c02 CHAR(20), c03 TEXT, c04 DOUBLE);
+Warnings:
+Note	1105	build_frm_image: Field data type info length: 12
+Note	1105	DBUG: [0] name='c01' type_info=''
+Note	1105	DBUG: [1] name='c02' type_info='char'
+Note	1105	DBUG: [2] name='c03' type_info='blob'
+Note	1105	DBUG: [3] name='c04' type_info=''
+DROP TABLE t1;
+SET SESSION debug_dbug="-d,frm_data_type_info_emulate";
+SET SESSION debug_dbug="-d,frm_data_type_info";

mysql-test/main/frm-debug.test

+--source include/have_debug.inc
+
+--echo #
+--echo # MDEV-20042 Implement EXTRA2_FIELD_DATA_TYPE_INFO in FRM
+--echo #
+
+# This should have empty EXTRA2_FIELD_DATA_TYPE_INFO
+SET SESSION debug_dbug="+d,frm_data_type_info";
+CREATE TABLE t1 (c01 INT, c02 CHAR(20), c03 TEXT, c04 DOUBLE);
+DROP TABLE t1;
+SET SESSION debug_dbug="-d,frm_data_type_info";
+
+# This should have non-empty EXTRA2_FIELD_DATA_TYPE_INFO
+SET SESSION debug_dbug="+d,frm_data_type_info";
+SET SESSION debug_dbug="+d,frm_data_type_info_emulate";
+CREATE TABLE t1 (c01 INT, c02 CHAR(20), c03 TEXT, c04 DOUBLE);
+DROP TABLE t1;
+SET SESSION debug_dbug="-d,frm_data_type_info_emulate";
+SET SESSION debug_dbug="-d,frm_data_type_info";

sql/sql_string.h

@@ -528,6 +528,10 @@ class Binary_string: public Static_binary_string
     q_append(s, size);
     return false;
   }
+  bool append(const LEX_CSTRING &s)
+  {
+    return append(s.str, s.length);
+  }
   bool append(const Binary_string &s)
   {
     return append(s.ptr(), s.length());
@@ -1001,6 +1005,15 @@ class StringBuffer : public String
 };
 
 
+template<size_t buff_sz>
+class BinaryStringBuffer : public Binary_string
+{
+  char buff[buff_sz];
+public:
+  BinaryStringBuffer() : Binary_string(buff, buff_sz) { length(0); }
+};
+
+
 class String_space: public String
 {
 public:

sql/sql_type.cc

@@ -8365,6 +8365,20 @@ const Name & Type_handler_timestamp_common::default_value() const
   return def;
 }
 
+/***************************************************************************/
+
+bool Type_handler::Column_definition_data_type_info_image(Binary_string *to,
+                                                   const Column_definition &def)
+                                                   const
+{
+  // Have *some* columns write type info (let's use string fields as an example)
+  DBUG_EXECUTE_IF("frm_data_type_info_emulate",
+                  if (cmp_type() == STRING_RESULT)
+                    return to->append(name().lex_cstring()););
+  return false;
+}
+
+
 /***************************************************************************/
 
 LEX_CSTRING Charset::collation_specific_name() const

sql/sql_type.h

@@ -3076,6 +3076,7 @@ class Name: private LEX_CSTRING
   }
   const char *ptr() const { return LEX_CSTRING::str; }
   uint length() const { return (uint) LEX_CSTRING::length; }
+  const LEX_CSTRING &lex_cstring() const { return *this; }
   bool eq(const LEX_CSTRING &other) const
   {
     return !my_strnncoll(system_charset_info,
@@ -3490,6 +3491,9 @@ class Type_handler
   {
     return 0;
   }
+  virtual bool Column_definition_data_type_info_image(Binary_string *to,
+                                                   const Column_definition &def)
+                                                   const;
   // Check if the implicit default value is Ok in the current sql_mode
   virtual bool validate_implicit_default_value(THD *thd,
                                                const Column_definition &def)

sql/table.cc

@@ -59,6 +59,7 @@ struct extra2_fields
   LEX_CUSTRING field_flags;
   LEX_CUSTRING system_period;
   LEX_CUSTRING application_period;
+  LEX_CUSTRING field_data_type_info;
   void reset()
   { bzero((void*)this, sizeof(*this)); }
 };
@@ -1508,6 +1509,12 @@ bool read_extra2(const uchar *frm_image, size_t len, extra2_fields *fields)
           fields->application_period.str= extra2;
           fields->application_period.length= length;
           break;
+        case EXTRA2_FIELD_DATA_TYPE_INFO:
+          if (fields->field_data_type_info.str)
+            DBUG_RETURN(true);
+          fields->field_data_type_info.str= extra2;
+          fields->field_data_type_info.length= length;
+          break;
         default:
           /* abort frm parsing if it's an unknown but important extra2 value */
           if (type >= EXTRA2_ENGINE_IMPORTANT)
@@ -1522,6 +1529,86 @@ bool read_extra2(const uchar *frm_image, size_t len, extra2_fields *fields)
 }
 
 
+class Field_data_type_info_array
+{
+public:
+  class Elem
+  {
+    LEX_CSTRING m_type_info;
+  public:
+    void set(const LEX_CSTRING &type_info)
+    {
+      m_type_info= type_info;
+    }
+    const LEX_CSTRING &type_info() const
+    {
+      return m_type_info;
+    }
+  };
+private:
+  Elem *m_array;
+  uint m_count;
+  bool alloc(MEM_ROOT *root, uint count)
+  {
+    DBUG_ASSERT(!m_array);
+    DBUG_ASSERT(!m_count);
+    size_t nbytes= sizeof(Elem) * count;
+    if (!(m_array= (Elem*) alloc_root(root, nbytes)))
+      return true;
+    m_count= count;
+    bzero((void*) m_array, nbytes);
+    return false;
+  }
+  static uint32 read_length(uchar **pos, const uchar *end)
+  {
+    ulonglong num= safe_net_field_length_ll(pos, end - *pos);
+    if (num > UINT_MAX32)
+      return 0;
+    return (uint32) num;
+  }
+  static bool read_string(LEX_CSTRING *to, uchar **pos, const uchar *end)
+  {
+    to->length= read_length(pos, end);
+    if (*pos + to->length > end)
+      return true; // Not enough data
+    to->str= (const char *) *pos;
+    *pos+= to->length;
+    return false;
+  }
+public:
+  Field_data_type_info_array()
+   :m_array(NULL), m_count(0)
+  { }
+  uint count() const
+  {
+    return m_count;
+  }
+  const Elem element(uint i) const
+  {
+    DBUG_ASSERT(i < m_count);
+    return m_array[i];
+  }
+  bool parse(MEM_ROOT *root, uint count, LEX_CUSTRING &image)
+  {
+    const uchar *pos= image.str;
+    const uchar *end= pos + image.length;
+    if (alloc(root, count))
+      return true;
+    for (uint i= 0; i < count && pos < end; i++)
+    {
+      LEX_CSTRING type_info;
+      uint fieldnr= read_length((uchar**) &pos, end);
+      if ((fieldnr == 0 && i > 0) || fieldnr >= count)
+        return true; // Bad data
+      if (read_string(&type_info, (uchar**) &pos, end) || type_info.length == 0)
+        return true; // Bad data
+      m_array[fieldnr].set(type_info);
+    }
+    return pos < end; // Error if some data is still left
+  }
+};
+
+
 /**
   Read data from a binary .frm file image into a TABLE_SHARE
 
@@ -1572,6 +1659,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
   uint ext_key_parts= 0;
   plugin_ref se_plugin= 0;
   bool vers_can_native= false;
+  Field_data_type_info_array field_data_type_info_array;
 
   MEM_ROOT *old_root= thd->mem_root;
   Virtual_column_info **table_check_constraints;
@@ -2111,6 +2199,11 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     status_var_increment(thd->status_var.feature_application_time_periods);
   }
 
+  if (extra2.field_data_type_info.length &&
+      field_data_type_info_array.parse(old_root, share->fields,
+                                       extra2.field_data_type_info))
+    goto err;
+
   for (i=0 ; i < share->fields; i++, strpos+=field_pack_length, field_ptr++)
   {
     uint interval_nr= 0, recpos;
@@ -2195,6 +2288,16 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
                                                              strpos,
                                                              &extra2.gis))
           goto err;
+
+        if (field_data_type_info_array.count())
+        {
+          DBUG_EXECUTE_IF("frm_data_type_info",
+            push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
+            ER_UNKNOWN_ERROR, "DBUG: [%u] name='%s' type_info='%.*s'",
+            i, share->fieldnames.type_names[i],
+            (uint) field_data_type_info_array.element(i).type_info().length,
+            field_data_type_info_array.element(i).type_info().str););
+        }
       }
 
       if (((uint) strpos[10]) & MYSQL57_GENERATED_FIELD)

sql/unireg.cc

@@ -171,6 +171,61 @@ static uint gis_field_options_image(uchar *buff,
 }
 
 
+class Field_data_type_info_image: public BinaryStringBuffer<512>
+{
+  static uchar *store_length(uchar *pos, ulonglong length)
+  {
+    return net_store_length(pos, length);
+  }
+  static uchar *store_string(uchar *pos, const LEX_CSTRING &str)
+  {
+    pos= store_length(pos, str.length);
+    memcpy(pos, str.str, str.length);
+    return pos + str.length;
+  }
+  static uint store_length_required_length(ulonglong length)
+  {
+    return net_length_size(length);
+  }
+public:
+  Field_data_type_info_image() { }
+  bool append(uint fieldnr, const Column_definition &def)
+  {
+    BinaryStringBuffer<64> type_info;
+    if (def.type_handler()->
+              Column_definition_data_type_info_image(&type_info, def) ||
+        type_info.length() > 0xFFFF/*Some reasonable limit*/)
+      return true; // Error
+    if (!type_info.length())
+      return false;
+    size_t need_length= store_length_required_length(fieldnr) +
+                        store_length_required_length(type_info.length()) +
+                        type_info.length();
+    if (reserve(need_length))
+      return true; // Error
+    uchar *pos= (uchar *) end();
+    pos= store_length(pos, fieldnr);
+    pos= store_string(pos, type_info.lex_cstring());
+    size_t new_length= (const char *) pos - ptr();
+    DBUG_ASSERT(new_length < alloced_length());
+    length((uint32) new_length);
+    return false;
+  }
+  bool append(List<Create_field> &fields)
+  {
+    uint fieldnr= 0;
+    Create_field *field;
+    List_iterator<Create_field> it(fields);
+    for (field= it++; field; field= it++, fieldnr++)
+    {
+      if (append(fieldnr, *field))
+        return true; // Error
+    }
+    return false;
+  }
+};
+
+
 /**
   Create a frm (table definition) file
 
@@ -210,6 +265,7 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING &table,
   uchar *frm_ptr, *pos;
   LEX_CUSTRING frm= {0,0};
   StringBuffer<MAX_FIELD_WIDTH> vcols;
+  Field_data_type_info_image field_data_type_info_image;
   DBUG_ENTER("build_frm_image");
 
  /* If fixed row records, we need one bit to check for deleted rows */
@@ -263,6 +319,22 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING &table,
   gis_extra2_len= gis_field_options_image(NULL, create_fields);
   DBUG_PRINT("info", ("Options length: %u", options_len));
 
+  if (field_data_type_info_image.append(create_fields))
+  {
+    my_printf_error(ER_CANT_CREATE_TABLE,
+                    "Cannot create table %`s: "
+                    "Building the field data type info image failed.",
+                    MYF(0), table.str);
+    DBUG_RETURN(frm);
+  }
+  DBUG_PRINT("info", ("Field data type info length: %u",
+                      (uint) field_data_type_info_image.length()));
+  DBUG_EXECUTE_IF("frm_data_type_info",
+                  push_warning_printf(thd, Sql_condition::WARN_LEVEL_NOTE,
+                  ER_UNKNOWN_ERROR,
+                  "build_frm_image: Field data type info length: %u",
+                  (uint) field_data_type_info_image.length()););
+
   if (validate_comment_length(thd, &create_info->comment, TABLE_COMMENT_MAXLEN,
                               ER_TOO_LONG_TABLE_COMMENT, table.str))
      DBUG_RETURN(frm);
@@ -308,6 +380,9 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING &table,
   if (gis_extra2_len)
     extra2_size+= 1 + extra2_str_size(gis_extra2_len);
 
+  if (field_data_type_info_image.length())
+    extra2_size+= 1 + extra2_str_size(field_data_type_info_image.length());
+
   if (create_info->versioned())
   {
     extra2_size+= 1 + extra2_str_size(2 * frm_fieldno_size);
@@ -380,6 +455,22 @@ LEX_CUSTRING build_frm_image(THD *thd, const LEX_CSTRING &table,
     pos+= gis_field_options_image(pos, create_fields);
   }
 
+  if (field_data_type_info_image.length())
+  {
+    if (field_data_type_info_image.length() > 0xFFFF)
+    {
+      my_printf_error(ER_CANT_CREATE_TABLE,
+                      "Cannot create table %`s: "
+                      "field data type info image is too large. "
+                      "Decrease the number of columns with "
+                      "extended data types.",
+                      MYF(0), table.str);
+      goto err;
+    }
+    *pos= EXTRA2_FIELD_DATA_TYPE_INFO;
+    pos= extra2_write_str(pos + 1, field_data_type_info_image.lex_cstring());
+  }
+
   // PERIOD
   if (create_info->period_info.is_set())
   {

sql/unireg.h

@@ -176,7 +176,8 @@ enum extra2_frm_value_type {
 #define EXTRA2_ENGINE_IMPORTANT 128
 
   EXTRA2_ENGINE_TABLEOPTS=128,
-  EXTRA2_FIELD_FLAGS=129
+  EXTRA2_FIELD_FLAGS=129,
+  EXTRA2_FIELD_DATA_TYPE_INFO=130
 };
 
 enum extra2_field_flags {