MDEV-12695 Add Column_definition::type_handler()

20 files changed, 202 insertions, 213 deletions

diff --git a/mysql-test/r/signal_code.result b/mysql-test/r/signal_code.result
index ca46f1d2079..32192251fe9 100644
--- a/mysql-test/r/signal_code.result
+++ b/mysql-test/r/signal_code.result
@@ -31,5 +31,5 @@ Pos	Instruction
 1	stmt 130 "SIGNAL foo SET MESSAGE_TEXT = "This i..."
 2	stmt 131 "RESIGNAL foo"
 3	stmt 131 "RESIGNAL foo SET MESSAGE_TEXT = "This..."
-4	freturn 3 0
+4	freturn int 0
 drop function signal_func;
diff --git a/mysql-test/r/sp-code.result b/mysql-test/r/sp-code.result
index f6a7fff5d4e..972c7ff5b16 100644
--- a/mysql-test/r/sp-code.result
+++ b/mysql-test/r/sp-code.result
@@ -11,7 +11,7 @@ returns int
 return 0;
 show function code almost_empty;
 Pos	Instruction
-0	freturn 3 0
+0	freturn int 0
 drop function almost_empty;
 create procedure code_sample(x int, out err int, out nulls int)
 begin
@@ -957,10 +957,10 @@ END
 $
 SHOW FUNCTION CODE testf_bug11763507;
 Pos	Instruction
-0	freturn 3 0
+0	freturn int 0
 SHOW FUNCTION CODE TESTF_bug11763507;
 Pos	Instruction
-0	freturn 3 0
+0	freturn int 0
 SHOW PROCEDURE CODE testp_bug11763507;
 Pos	Instruction
 0	stmt 0 "SELECT "PROCEDURE testp_bug11763507""
diff --git a/mysql-test/suite/compat/oracle/r/sp-code.result b/mysql-test/suite/compat/oracle/r/sp-code.result
index 9d1b2e3f3f5..1087e10552b 100644
--- a/mysql-test/suite/compat/oracle/r/sp-code.result
+++ b/mysql-test/suite/compat/oracle/r/sp-code.result
@@ -11,7 +11,7 @@ END;
 /
 SHOW FUNCTION CODE f1;
 Pos	Instruction
-0	freturn 3 10
+0	freturn int 10
 SELECT f1();
 f1()
 10
@@ -418,7 +418,7 @@ Pos	Instruction
 1	set i@0 i@0 + 1
 2	jump_if_not 1(1) i@0 >= 5
 3	jump 4
-4	freturn 3 i@0
+4	freturn int i@0
 SELECT f1() FROM DUAL;
 f1()
 5
@@ -440,7 +440,7 @@ Pos	Instruction
 1	set i@0 i@0 + 1
 2	jump_if_not 1(0) i@0 >= 5
 3	jump 4
-4	freturn 3 i@0
+4	freturn int i@0
 SELECT f1() FROM DUAL;
 f1()
 5
@@ -474,7 +474,7 @@ Pos	Instruction
 7	hreturn 0 8
 8	hpop 1
 9	jump 5
-10	freturn 3 i@0
+10	freturn int i@0
 SELECT f1() FROM DUAL;
 f1()
 5
@@ -586,7 +586,7 @@ Pos	Instruction
 6	jump 9
 7	set i@3 i@3 + 1
 8	jump 3
-9	freturn 3 total@2
+9	freturn int total@2
 SELECT f1(3, 100) FROM DUAL;
 f1(3, 100)
 6
@@ -619,7 +619,7 @@ Pos	Instruction
 6	jump 9
 7	set i@3 i@3 + -1
 8	jump 3
-9	freturn 3 total@2
+9	freturn int total@2
 SELECT f1(3, 100) FROM DUAL;
 f1(3, 100)
 6
@@ -666,7 +666,7 @@ Pos	Instruction
 14	jump 7
 15	set ia@5 ia@5 + 1
 16	jump 3
-17	freturn 3 total@4
+17	freturn int total@4
 SELECT f1(2, 1, 2, 2) FROM DUAL;
 f1(2, 1, 2, 2)
 1001
@@ -707,7 +707,7 @@ Pos	Instruction
 8	set total@1 total@1 + 1
 9	set i@2 i@2 + 1
 10	jump 3
-11	freturn 3 total@1
+11	freturn int total@1
 SELECT f1(3), f1(4), f1(5), f1(6) FROM DUAL;
 f1(3)	f1(4)	f1(5)	f1(6)
 3003	4004	5004	6005
@@ -749,7 +749,7 @@ Pos	Instruction
 13	jump 6
 14	set i@2 i@2 + 1
 15	jump 3
-16	freturn 3 total@1
+16	freturn int total@1
 SELECT f1(3), f1(4), f1(5), f1(6) FROM DUAL;
 f1(3)	f1(4)	f1(5)	f1(6)
 6006	8008	9008	11010
@@ -792,7 +792,7 @@ Pos	Instruction
 13	jump 6
 14	set j@2 j@2 + 1
 15	jump 3
-16	freturn 3 total@1
+16	freturn int total@1
 SELECT f1(3), f1(4), f1(5), f1(6) FROM DUAL;
 f1(3)	f1(4)	f1(5)	f1(6)
 6006	8008	10008	12010
@@ -822,7 +822,7 @@ Pos	Instruction
 7	set total@1 total@1 + 1
 8	set i@2 i@2 + 1
 9	jump 3
-10	freturn 3 total@1
+10	freturn int total@1
 SELECT f1(3), f1(4), f1(5), f1(6) FROM DUAL;
 f1(3)	f1(4)	f1(5)	f1(6)
 3	4	4	5
@@ -902,7 +902,7 @@ SHOW FUNCTION CODE f1;
 Pos	Instruction
 0	set a@0 NULL
 1	set a.b@0[1] 200
-2	freturn 3 a.b@0[1]
+2	freturn int a.b@0[1]
 SELECT f1();
 f1()
 200
diff --git a/sql/field.cc b/sql/field.cc
index 8b4aac93a9c..ebb7e1fa479 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -999,37 +999,6 @@ CPP_UNNAMED_NS_END
   Static help functions
 *****************************************************************************/
 
-/**
-  Check whether a field type can be partially indexed by a key.
-
-  This is a static method, rather than a virtual function, because we need
-  to check the type of a non-Field in mysql_alter_table().
-
-  @param type  field type
-
-  @retval
-    TRUE  Type can have a prefixed key
-  @retval
-    FALSE Type can not have a prefixed key
-*/
-
-bool Field::type_can_have_key_part(enum enum_field_types type)
-{
-  switch (type) {
-  case MYSQL_TYPE_VARCHAR:
-  case MYSQL_TYPE_TINY_BLOB:
-  case MYSQL_TYPE_MEDIUM_BLOB:
-  case MYSQL_TYPE_LONG_BLOB:
-  case MYSQL_TYPE_BLOB:
-  case MYSQL_TYPE_VAR_STRING:
-  case MYSQL_TYPE_STRING:
-  case MYSQL_TYPE_GEOMETRY:
-    return TRUE;
-  default:
-    return FALSE;
-  }
-}
-
 
 void Field::make_sort_key(uchar *buff,uint length)
 {
@@ -3294,7 +3263,7 @@ bool Field_new_decimal::compatible_field_size(uint field_metadata,
 
 uint Field_new_decimal::is_equal(Create_field *new_field)
 {
-  return ((new_field->sql_type == real_type()) &&
+  return ((new_field->type_handler() == type_handler()) &&
           ((new_field->flags & UNSIGNED_FLAG) == 
            (uint) (flags & UNSIGNED_FLAG)) &&
           ((new_field->flags & AUTO_INCREMENT_FLAG) ==
@@ -5378,7 +5347,7 @@ my_time_t Field_timestampf::get_timestamp(const uchar *pos,
 /*************************************************************/
 uint Field_temporal::is_equal(Create_field *new_field)
 {
-  return new_field->sql_type == real_type() &&
+  return new_field->type_handler() == type_handler() &&
          new_field->length == max_display_length();
 }
 
@@ -6851,15 +6820,15 @@ int Field_str::store(double nr)
 
 uint Field::is_equal(Create_field *new_field)
 {
-  return (new_field->sql_type == real_type());
+  return new_field->type_handler() == type_handler();
 }
 
 
 uint Field_str::is_equal(Create_field *new_field)
 {
-  return ((new_field->sql_type == real_type()) &&
-	  new_field->charset == field_charset &&
-	  new_field->length == max_display_length());
+  return new_field->type_handler() == type_handler() &&
+         new_field->charset == field_charset &&
+         new_field->length == max_display_length();
 }
 
 
@@ -7717,7 +7686,7 @@ Field *Field_varstring::new_key_field(MEM_ROOT *root, TABLE *new_table,
 
 uint Field_varstring::is_equal(Create_field *new_field)
 {
-  if (new_field->sql_type == real_type() &&
+  if (new_field->type_handler() == type_handler() &&
       new_field->charset == field_charset)
   {
     if (new_field->length == max_display_length())
@@ -8292,9 +8261,9 @@ uint Field_blob::max_packed_col_length(uint max_length)
 
 uint Field_blob::is_equal(Create_field *new_field)
 {
-  return ((new_field->sql_type == get_blob_type_from_length(max_data_length()))
-          && new_field->charset == field_charset &&
-          new_field->pack_length == pack_length());
+  return new_field->type_handler() == type_handler() &&
+         new_field->charset == field_charset &&
+         new_field->pack_length == pack_length();
 }
 
 
@@ -8321,7 +8290,7 @@ uint gis_field_options_image(uchar *buff, List<Create_field> &create_fields)
   Create_field *field;
   while ((field= it++))
   {
-    if (field->sql_type != MYSQL_TYPE_GEOMETRY)
+    if (field->real_field_type() != MYSQL_TYPE_GEOMETRY)
       continue;
    if (buff)
    {
@@ -8514,7 +8483,7 @@ Field::geometry_type Field_geom::geometry_type_merge(geometry_type a,
 
 uint Field_geom::is_equal(Create_field *new_field)
 {
-  return new_field->sql_type == MYSQL_TYPE_GEOMETRY &&
+  return new_field->type_handler() == type_handler() &&
          /*
            - Allow ALTER..INPLACE to supertype (GEOMETRY),
              e.g. POINT to GEOMETRY or POLYGON to GEOMETRY.
@@ -8941,7 +8910,7 @@ uint Field_enum::is_equal(Create_field *new_field)
     The fields are compatible if they have the same flags,
     type, charset and have the same underlying length.
   */
-  if (new_field->sql_type != real_type() ||
+  if (new_field->type_handler() != type_handler() ||
       new_field->charset != field_charset ||
       new_field->pack_length != pack_length())
     return IS_EQUAL_NO;
@@ -9006,7 +8975,7 @@ bool Field_num::eq_def(const Field *field) const
 
 uint Field_num::is_equal(Create_field *new_field)
 {
-  return ((new_field->sql_type == real_type()) &&
+  return ((new_field->type_handler() == type_handler()) &&
           ((new_field->flags & UNSIGNED_FLAG) == 
            (uint) (flags & UNSIGNED_FLAG)) &&
 	  ((new_field->flags & AUTO_INCREMENT_FLAG) ==
@@ -9160,8 +9129,8 @@ Field *Field_bit::new_key_field(MEM_ROOT *root, TABLE *new_table,
 
 uint Field_bit::is_equal(Create_field *new_field) 
 {
-  return (new_field->sql_type == real_type() &&
-          new_field->length == max_display_length());
+  return new_field->type_handler() == type_handler() &&
+         new_field->length == max_display_length();
 }
 
                        
@@ -9699,7 +9668,7 @@ bool Column_definition::create_interval_from_interval_list(MEM_ROOT *mem_root,
     value.length= charset->cset->lengthsp(charset, value.str, value.length);
     ((char*) value.str)[value.length]= '\0';
 
-    if (sql_type == MYSQL_TYPE_SET)
+    if (real_field_type() == MYSQL_TYPE_SET)
     {
       if (charset->coll->instr(charset, value.str, value.length,
                                comma_buf, comma_length, NULL, 0))
@@ -9726,7 +9695,8 @@ bool Column_definition::prepare_interval_field(MEM_ROOT *mem_root,
                                                bool reuse_interval_list_values)
 {
   DBUG_ENTER("Column_definition::prepare_interval_field");
-  DBUG_ASSERT(sql_type == MYSQL_TYPE_ENUM || sql_type == MYSQL_TYPE_SET);
+  DBUG_ASSERT(real_field_type() == MYSQL_TYPE_ENUM ||
+              real_field_type() == MYSQL_TYPE_SET);
   /*
     Interval values are either in "interval" or in "interval_list",
     but not in both at the same time, and are not empty at the same time.
@@ -9779,12 +9749,12 @@ bool Column_definition::prepare_interval_field(MEM_ROOT *mem_root,
 void Column_definition::set_attributes(const Lex_field_type_st &type,
                                        CHARSET_INFO *cs)
 {
-  DBUG_ASSERT(sql_type == MYSQL_TYPE_NULL);
+  DBUG_ASSERT(type_handler() == &type_handler_null);
   DBUG_ASSERT(charset == &my_charset_bin || charset == NULL);
   DBUG_ASSERT(length == 0);
   DBUG_ASSERT(decimals == 0);
 
-  sql_type= type.field_type();
+  set_handler_by_real_type(type.field_type());
   charset= cs;
 
   if (type.length())
@@ -9806,7 +9776,7 @@ void Column_definition::set_attributes(const Lex_field_type_st &type,
 
 void Column_definition::create_length_to_internal_length(void)
 {
-  switch (sql_type) {
+  switch (real_field_type()) {
   case MYSQL_TYPE_TINY_BLOB:
   case MYSQL_TYPE_MEDIUM_BLOB:
   case MYSQL_TYPE_LONG_BLOB:
@@ -9817,7 +9787,7 @@ void Column_definition::create_length_to_internal_length(void)
   case MYSQL_TYPE_VARCHAR:
     length*= charset->mbmaxlen;
     key_length= length;
-    pack_length= calc_pack_length(sql_type, length);
+    pack_length= calc_pack_length(real_field_type(), length);
     break;
   case MYSQL_TYPE_ENUM:
   case MYSQL_TYPE_SET:
@@ -9846,7 +9816,7 @@ void Column_definition::create_length_to_internal_length(void)
 				 decimals);
     break;
   default:
-    key_length= pack_length= calc_pack_length(sql_type, length);
+    key_length= pack_length= calc_pack_length(real_field_type(), length);
     break;
   }
 }
@@ -9901,7 +9871,7 @@ bool Column_definition::check(THD *thd)
   if (vcol_info)
   {
     DBUG_ASSERT(vcol_info->expr);
-    vcol_info->set_field_type(sql_type);
+    vcol_info->set_field_type(real_field_type());
     if (check_expression(vcol_info, &field_name, vcol_info->stored_in_db
                          ? VCOL_GENERATED_STORED : VCOL_GENERATED_VIRTUAL))
       DBUG_RETURN(TRUE);
@@ -9939,7 +9909,7 @@ bool Column_definition::check(THD *thd)
   }
 
   if (default_value && !default_value->expr->basic_const_item() &&
-      mysql_type_to_time_type(sql_type) == MYSQL_TIMESTAMP_DATETIME &&
+      mysql_timestamp_type() == MYSQL_TIMESTAMP_DATETIME &&
       default_value->expr->type() == Item::FUNC_ITEM)
   {
     /*
@@ -9957,7 +9927,7 @@ bool Column_definition::check(THD *thd)
 
   if (on_update)
   {
-    if (mysql_type_to_time_type(sql_type) != MYSQL_TIMESTAMP_DATETIME ||
+    if (mysql_timestamp_type() != MYSQL_TIMESTAMP_DATETIME ||
         on_update->decimals < length)
     {
       my_error(ER_INVALID_ON_UPDATE, MYF(0), field_name.str);
@@ -9971,7 +9941,7 @@ bool Column_definition::check(THD *thd)
 
   sign_len= flags & UNSIGNED_FLAG ? 0 : 1;
 
-  switch (sql_type) {
+  switch (real_field_type()) {
   case MYSQL_TYPE_TINY:
     if (!length)
       length= MAX_TINYINT_WIDTH+sign_len;
@@ -10098,7 +10068,7 @@ bool Column_definition::check(THD *thd)
     break;
   case MYSQL_TYPE_DATE:
     /* We don't support creation of MYSQL_TYPE_DATE anymore */
-    sql_type= MYSQL_TYPE_NEWDATE;
+    set_handler(&type_handler_newdate);
     /* fall trough */
   case MYSQL_TYPE_NEWDATE:
     length= MAX_DATE_WIDTH;
@@ -10165,12 +10135,13 @@ bool Column_definition::check(THD *thd)
       explicit_defaults_for_timestamp is not set.
     */
     if (opt_explicit_defaults_for_timestamp ||
-        !is_timestamp_type(sql_type))
+        !is_timestamp_type())
     {
       flags|= NO_DEFAULT_VALUE_FLAG;
     }
   }
 
+  enum_field_types sql_type= real_field_type();
   if (!(flags & BLOB_FLAG) &&
       ((length > max_field_charlength &&
         sql_type != MYSQL_TYPE_VARCHAR) ||
@@ -10295,7 +10266,7 @@ Field *make_field(TABLE_SHARE *share,
                   uchar *ptr, uint32 field_length,
 		  uchar *null_pos, uchar null_bit,
 		  uint pack_flag,
-		  enum_field_types field_type,
+		  const Type_handler *handler,
 		  CHARSET_INFO *field_charset,
 		  Field::geometry_type geom_type, uint srid,
 		  Field::utype unireg_check,
@@ -10305,7 +10276,7 @@ Field *make_field(TABLE_SHARE *share,
   uchar *UNINIT_VAR(bit_ptr);
   uchar UNINIT_VAR(bit_offset);
 
-  if (field_type == MYSQL_TYPE_BIT && !f_bit_as_char(pack_flag))
+  if (handler->real_field_type() == MYSQL_TYPE_BIT && !f_bit_as_char(pack_flag))
   {
     bit_ptr= null_pos;
     bit_offset= null_bit;
@@ -10326,8 +10297,8 @@ Field *make_field(TABLE_SHARE *share,
     null_bit= ((uchar) 1) << null_bit;
   }
 
-  DBUG_PRINT("debug", ("field_type: %d, field_length: %u, interval: %p, pack_flag: %s%s%s%s%s",
-                       field_type, field_length, interval,
+  DBUG_PRINT("debug", ("field_type: %s, field_length: %u, interval: %p, pack_flag: %s%s%s%s%s",
+                       handler->name().ptr(), field_length, interval,
                        FLAGSTR(pack_flag, FIELDFLAG_BINARY),
                        FLAGSTR(pack_flag, FIELDFLAG_INTERVAL),
                        FLAGSTR(pack_flag, FIELDFLAG_NUMBER),
@@ -10338,6 +10309,7 @@ Field *make_field(TABLE_SHARE *share,
   {
     if (!f_is_packed(pack_flag))
     {
+      enum_field_types field_type= handler->real_field_type();
       if (field_type == MYSQL_TYPE_STRING ||
           field_type == MYSQL_TYPE_DECIMAL ||   // 3.23 or 4.0 string
           field_type == MYSQL_TYPE_VAR_STRING)
@@ -10390,7 +10362,7 @@ Field *make_field(TABLE_SHARE *share,
     }
   }
 
-  switch (field_type) {
+  switch (handler->real_field_type()) {
   case MYSQL_TYPE_DECIMAL:
     return new (mem_root)
       Field_decimal(ptr,field_length,null_pos,null_bit,
@@ -10548,7 +10520,7 @@ Column_definition::Column_definition(THD *thd, Field *old_field,
   unireg_check=old_field->unireg_check;
   pack_length=old_field->pack_length();
   key_length= old_field->key_length();
-  sql_type=   old_field->real_type();
+  set_handler(old_field->type_handler());
   charset=    old_field->charset();		// May be NULL ptr
   comment=    old_field->comment;
   decimals=   old_field->decimals();
@@ -10558,21 +10530,18 @@ Column_definition::Column_definition(THD *thd, Field *old_field,
   option_list= old_field->option_list;
   pack_flag= 0;
 
-  switch (sql_type) {
+  switch (real_field_type()) {
+  case MYSQL_TYPE_TINY_BLOB:
   case MYSQL_TYPE_BLOB:
-    switch (pack_length - portable_sizeof_char_ptr) {
-    case  1: sql_type= MYSQL_TYPE_TINY_BLOB; break;
-    case  2: sql_type= MYSQL_TYPE_BLOB; break;
-    case  3: sql_type= MYSQL_TYPE_MEDIUM_BLOB; break;
-    default: sql_type= MYSQL_TYPE_LONG_BLOB; break;
-    }
+  case MYSQL_TYPE_MEDIUM_BLOB:
+  case MYSQL_TYPE_LONG_BLOB:
     length/= charset->mbmaxlen;
     key_length/= charset->mbmaxlen;
     break;
   case MYSQL_TYPE_STRING:
     /* Change CHAR -> VARCHAR if dynamic record length */
     if (old_field->type() == MYSQL_TYPE_VAR_STRING)
-      sql_type= MYSQL_TYPE_VARCHAR;
+      set_handler(&type_handler_varchar);
     /* fall through */
 
   case MYSQL_TYPE_ENUM:
diff --git a/sql/field.h b/sql/field.h
index 5b7b08ec37a..ccc116f7a25 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -460,16 +460,6 @@ inline bool is_temporal_type_with_date(enum_field_types type)
 
 
 /**
-   Recognizer for concrete data type (called real_type for some reason),
-   returning true if it is one of the TIMESTAMP types.
-*/
-inline bool is_timestamp_type(enum_field_types type)
-{
-  return type == MYSQL_TYPE_TIMESTAMP || type == MYSQL_TYPE_TIMESTAMP2;
-}
-
-
-/**
   Convert temporal real types as retuned by field->real_type()
   to field type as returned by field->type().
   
@@ -838,7 +828,6 @@ public:
   {
     return type_handler()->cmp_type();
   }
-  static bool type_can_have_key_part(enum_field_types);
   static enum_field_types field_type_merge(enum_field_types, enum_field_types);
   virtual bool eq(Field *field)
   {
@@ -3765,7 +3754,7 @@ extern const LEX_CSTRING null_clex_str;
 Field *make_field(TABLE_SHARE *share, MEM_ROOT *mem_root,
                   uchar *ptr, uint32 field_length,
                   uchar *null_pos, uchar null_bit,
-                  uint pack_flag, enum_field_types field_type,
+                  uint pack_flag, const Type_handler *handler,
                   CHARSET_INFO *cs,
                   Field::geometry_type geom_type, uint srid,
                   Field::utype unireg_check,
@@ -3774,7 +3763,8 @@ Field *make_field(TABLE_SHARE *share, MEM_ROOT *mem_root,
 /*
   Create field class for CREATE TABLE
 */
-class Column_definition: public Sql_alloc
+class Column_definition: public Sql_alloc,
+                         public Type_handler_hybrid_field_type
 {
   /**
     Create "interval" from "interval_list".
@@ -3814,11 +3804,11 @@ class Column_definition: public Sql_alloc
       set_if_bigger(*max_length, (uint32)length);
     }
   }
+  const Type_handler *field_type() const; // Prevent using this
 public:
   LEX_CSTRING field_name;
   LEX_CSTRING comment;			// Comment for field
   Item *on_update;		        // ON UPDATE NOW()
-  enum	enum_field_types sql_type;
   /*
     At various stages in execution this can be length of field in bytes or
     max number of characters. 
@@ -3850,9 +3840,10 @@ public:
     *default_value,                  // Default value
     *check_constraint;               // Check constraint
 
-  Column_definition():
+  Column_definition()
+   :Type_handler_hybrid_field_type(&type_handler_null),
     comment(null_clex_str),
-    on_update(NULL), sql_type(MYSQL_TYPE_NULL), length(0), decimals(0),
+    on_update(NULL), length(0), decimals(0),
     flags(0), pack_length(0), key_length(0), unireg_check(Field::NONE),
     interval(0), charset(&my_charset_bin),
     srid(0), geom_type(Field::GEOM_GEOMETRY),
@@ -3861,20 +3852,6 @@ public:
   {
     interval_list.empty();
   }
-
-  Column_definition(LEX_CSTRING *name, enum_field_types type):
-    field_name(*name),
-    comment(null_clex_str),
-    on_update(NULL), sql_type(type), length(0), decimals(0),
-    flags(0), pack_length(0), key_length(0), unireg_check(Field::NONE),
-    interval(0), charset(&my_charset_bin),
-    srid(0), geom_type(Field::GEOM_GEOMETRY),
-    option_list(NULL), pack_flag(0),
-    vcol_info(0), default_value(0), check_constraint(0)
-  {
-    interval_list.empty();
-  }
-
   Column_definition(THD *thd, Field *field, Field *orig_field);
   void set_attributes(const Lex_field_type_st &type, CHARSET_INFO *cs);
   void create_length_to_internal_length(void);
@@ -3896,7 +3873,7 @@ public:
   void prepare_interval_field_calc_length()
   {
     uint32 field_length, dummy;
-    if (sql_type == MYSQL_TYPE_SET)
+    if (real_field_type() == MYSQL_TYPE_SET)
     {
       calculate_interval_lengths(&dummy, &field_length);
       length= field_length + (interval->count - 1);
@@ -3942,7 +3919,7 @@ public:
   {
     return ::make_field(share, mem_root, ptr,
                         (uint32)length, null_pos, null_bit,
-                        pack_flag, sql_type, charset,
+                        pack_flag, type_handler(), charset,
                         geom_type, srid, unireg_check, interval,
                         field_name_arg);
   }
@@ -4042,7 +4019,7 @@ public:
   {
     return is_row() || is_table_rowtype_ref() || is_cursor_rowtype_ref() ?
            &type_handler_row :
-           Type_handler::get_handler_by_field_type(sql_type);
+           Type_handler_hybrid_field_type::type_handler();
   }
   bool is_column_type_ref() const { return m_column_type_ref != 0; }
   bool is_table_rowtype_ref() const { return m_table_rowtype_ref != 0; }
diff --git a/sql/sp_head.cc b/sql/sp_head.cc
index fc4c05a670b..12b956623f8 100644
--- a/sql/sp_head.cc
+++ b/sql/sp_head.cc
@@ -3784,7 +3784,7 @@ sp_instr_freturn::print(String *str)
   if (str->reserve(1024+8+32)) // Add some for the expr. too
     return;
   str->qs_append(STRING_WITH_LEN("freturn "));
-  str->qs_append((uint)m_type);
+  str->qs_append(m_type_handler->name().ptr());
   str->qs_append(' ');
   m_value->print(str, enum_query_type(QT_ORDINARY |
                                       QT_ITEM_ORIGINAL_FUNC_NULLIF));
diff --git a/sql/sp_head.h b/sql/sp_head.h
index bb516187a57..e3c9022609b 100644
--- a/sql/sp_head.h
+++ b/sql/sp_head.h
@@ -1430,8 +1430,8 @@ class sp_instr_freturn : public sp_instr
 public:
 
   sp_instr_freturn(uint ip, sp_pcontext *ctx,
-		   Item *val, enum enum_field_types type_arg, LEX *lex)
-    : sp_instr(ip, ctx), m_value(val), m_type(type_arg),
+		   Item *val, const Type_handler *handler, LEX *lex)
+    : sp_instr(ip, ctx), m_value(val), m_type_handler(handler),
       m_lex_keeper(lex, TRUE)
   {}
 
@@ -1453,7 +1453,7 @@ public:
 protected:
 
   Item *m_value;
-  enum enum_field_types m_type;
+  const Type_handler *m_type_handler;
   sp_lex_keeper m_lex_keeper;
 
 }; // class sp_instr_freturn : public sp_instr
diff --git a/sql/sp_pcontext.h b/sql/sp_pcontext.h
index 841ea8526a5..215ebbe5f77 100644
--- a/sql/sp_pcontext.h
+++ b/sql/sp_pcontext.h
@@ -60,7 +60,7 @@ public:
   Spvar_definition field_def;
 
   /// Field-type of the SP-variable.
-  enum_field_types sql_type() const { return field_def.sql_type; }
+  enum_field_types sql_type() const { return field_def.real_field_type(); }
 
   const Type_handler *type_handler() const { return field_def.type_handler(); }
 
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index dad438d1013..3c8f088d196 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -5406,7 +5406,7 @@ sp_variable *LEX::sp_add_for_loop_variable(THD *thd, const LEX_CSTRING *name,
   sp_variable *spvar= spcont->add_variable(thd, name);
   spcont->declare_var_boundary(1);
   spvar->field_def.field_name= spvar->name;
-  spvar->field_def.sql_type= MYSQL_TYPE_LONGLONG;
+  spvar->field_def.set_handler(&type_handler_longlong);
   /*
     The below is a simplified version of what
     Column_definition::prepare_create_field() does for a LONGLONG field.
@@ -6363,7 +6363,8 @@ Item_splocal *LEX::create_item_spvar_row_field(THD *thd,
     if (!(item= new (thd->mem_root)
                 Item_splocal_row_field(thd, a, b,
                                        spv->offset, row_field_offset,
-                                       def->sql_type, pos_in_q, length_in_q)))
+                                       def->real_field_type(),
+                                       pos_in_q, length_in_q)))
       return NULL;
   }
 #ifndef DBUG_OFF
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 053b864ed68..0f093975a37 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -2214,7 +2214,7 @@ static int add_column_list_values(File fptr, partition_info *part_info,
             my_error(ER_FIELD_NOT_FOUND_PART_ERROR, MYF(0));
             return 1;
           }
-          if (check_part_field(sql_field->sql_type,
+          if (check_part_field(sql_field->real_field_type(),
                                sql_field->field_name.str,
                                &result_type,
                                &need_cs_check))
diff --git a/sql/sql_sequence.cc b/sql/sql_sequence.cc
index 19ba20d78cd..98912691904 100644
--- a/sql/sql_sequence.cc
+++ b/sql/sql_sequence.cc
@@ -27,7 +27,7 @@ struct Field_definition
 {
   const char *field_name;
   uint length;
-  enum enum_field_types sql_type;
+  const Type_handler *type_handler;
   LEX_CSTRING comment;
   ulong flags;
 };
@@ -45,19 +45,19 @@ struct Field_definition
 
 static Field_definition sequence_structure[]=
 {
-  {"next_value", 21, MYSQL_TYPE_LONGLONG, {STRING_WITH_LEN("next not cached value")},
+  {"next_value", 21, &type_handler_longlong, {STRING_WITH_LEN("next not cached value")},
    FL},
-  {"min_value", 21, MYSQL_TYPE_LONGLONG, {STRING_WITH_LEN("min value")}, FL},
-  {"max_value", 21, MYSQL_TYPE_LONGLONG, {STRING_WITH_LEN("max value")}, FL},
-  {"start", 21, MYSQL_TYPE_LONGLONG, {STRING_WITH_LEN("start value")},  FL},
-  {"increment", 21, MYSQL_TYPE_LONGLONG,
+  {"min_value", 21, &type_handler_longlong, {STRING_WITH_LEN("min value")}, FL},
+  {"max_value", 21, &type_handler_longlong, {STRING_WITH_LEN("max value")}, FL},
+  {"start", 21, &type_handler_longlong, {STRING_WITH_LEN("start value")},  FL},
+  {"increment", 21, &type_handler_longlong,
    {C_STRING_WITH_LEN("increment value")}, FL},
-  {"cache", 21, MYSQL_TYPE_LONGLONG, {STRING_WITH_LEN("cache size")}, FL},
-  {"cycle", 1, MYSQL_TYPE_TINY, {STRING_WITH_LEN("cycle state")},
+  {"cache", 21, &type_handler_longlong, {STRING_WITH_LEN("cache size")}, FL},
+  {"cycle", 1, &type_handler_tiny, {STRING_WITH_LEN("cycle state")},
    FL | UNSIGNED_FLAG },
-  {"round", 21, MYSQL_TYPE_LONGLONG,
+  {"round", 21, &type_handler_longlong,
    {STRING_WITH_LEN("How many cycles has been done")}, FL},
-  {NULL, 0, MYSQL_TYPE_LONGLONG, {STRING_WITH_LEN("")}, 0}
+  {NULL, 0, &type_handler_longlong, {STRING_WITH_LEN("")}, 0}
 };
 
 #undef FL
@@ -196,7 +196,7 @@ bool check_sequence_fields(LEX *lex, List<Create_field> *fields)
     if (my_strcasecmp(system_charset_info, field_def->field_name,
                       field->field_name.str) ||
         field->flags != field_def->flags ||
-        field->sql_type != field_def->sql_type)
+        field->type_handler() != field_def->type_handler)
     {
       reason= field->field_name.str;
       goto err;
@@ -235,7 +235,7 @@ bool prepare_sequence_fields(THD *thd, List<Create_field> *fields)
       DBUG_RETURN(TRUE); /* purify inspected */
 
     new_field->field_name=  field_name;
-    new_field->sql_type=    field_info->sql_type;
+    new_field->set_handler(field_info->type_handler);
     new_field->length=      field_info->length;
     new_field->char_length= field_info->length;
     new_field->comment=     field_info->comment;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 76755a028ed..813c74b60b2 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2869,7 +2869,7 @@ bool Column_definition::prepare_create_field(uint *blob_columns,
   */
   DBUG_ASSERT(charset);
 
-  switch (sql_type) {
+  switch (real_field_type()) {
   case MYSQL_TYPE_BLOB:
   case MYSQL_TYPE_MEDIUM_BLOB:
   case MYSQL_TYPE_TINY_BLOB:
@@ -2905,8 +2905,8 @@ bool Column_definition::prepare_create_field(uint *blob_columns,
     if (table_flags & HA_NO_VARCHAR)
     {
       /* convert VARCHAR to CHAR because handler is not yet up to date */
-      sql_type= MYSQL_TYPE_VAR_STRING;
-      pack_length= calc_pack_length(sql_type, (uint) length);
+      set_handler(&type_handler_var_string);
+      pack_length= calc_pack_length(real_field_type(), (uint) length);
       if ((length / charset->mbmaxlen) > MAX_FIELD_CHARLENGTH)
       {
         my_error(ER_TOO_BIG_FIELDLENGTH, MYF(0), field_name.str,
@@ -2950,7 +2950,7 @@ bool Column_definition::prepare_create_field(uint *blob_columns,
   case MYSQL_TYPE_TIME2:
   case MYSQL_TYPE_DATETIME2:
   case MYSQL_TYPE_NULL:
-    pack_flag= f_settype((uint) sql_type);
+    pack_flag= f_settype((uint) real_field_type());
     break;
   case MYSQL_TYPE_BIT:
     /* 
@@ -2981,7 +2981,7 @@ bool Column_definition::prepare_create_field(uint *blob_columns,
     pack_flag= (FIELDFLAG_NUMBER |
                 (flags & UNSIGNED_FLAG ? 0 : FIELDFLAG_DECIMAL) |
                 (flags & ZEROFILL_FLAG ? FIELDFLAG_ZEROFILL : 0) |
-                f_settype((uint) sql_type) |
+                f_settype((uint) real_field_type()) |
                 (decimals_orig << FIELDFLAG_DEC_SHIFT));
     break;
   }
@@ -3044,7 +3044,7 @@ void promote_first_timestamp_column(List<Create_field> *column_definitions)
 
   while ((column_definition= it++) != NULL)
   {
-    if (is_timestamp_type(column_definition->sql_type) ||    // TIMESTAMP
+    if (column_definition->is_timestamp_type() ||    // TIMESTAMP
         column_definition->unireg_check == Field::TIMESTAMP_OLD_FIELD) // Legacy
     {
       if ((column_definition->flags & NOT_NULL_FLAG) != 0 && // NOT NULL,
@@ -3229,8 +3229,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 	!(sql_field->charset= find_bin_collation(sql_field->charset)))
       DBUG_RETURN(TRUE);
 
-    if (sql_field->sql_type == MYSQL_TYPE_SET ||
-        sql_field->sql_type == MYSQL_TYPE_ENUM)
+    if (sql_field->real_field_type() == MYSQL_TYPE_SET ||
+        sql_field->real_field_type() == MYSQL_TYPE_ENUM)
     {
       /*
         Create the typelib in runtime memory - we will free the
@@ -3244,7 +3244,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         DBUG_RETURN(true); // E.g. wrong values with commas: SET('a,b')
     }
 
-    if (sql_field->sql_type == MYSQL_TYPE_BIT)
+    if (sql_field->real_field_type() == MYSQL_TYPE_BIT)
     { 
       sql_field->pack_flag= FIELDFLAG_NUMBER;
       if (file->ha_table_flags() & HA_CAN_BIT_FIELD)
@@ -3265,14 +3265,14 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     if (sql_field->default_value &&
         sql_field->default_value->expr->basic_const_item() &&
         save_cs != sql_field->default_value->expr->collation.collation &&
-        (sql_field->sql_type == MYSQL_TYPE_VAR_STRING ||
-         sql_field->sql_type == MYSQL_TYPE_STRING ||
-         sql_field->sql_type == MYSQL_TYPE_SET ||
-         sql_field->sql_type == MYSQL_TYPE_TINY_BLOB ||
-         sql_field->sql_type == MYSQL_TYPE_MEDIUM_BLOB ||
-         sql_field->sql_type == MYSQL_TYPE_LONG_BLOB ||
-         sql_field->sql_type == MYSQL_TYPE_BLOB ||
-         sql_field->sql_type == MYSQL_TYPE_ENUM))
+        (sql_field->real_field_type() == MYSQL_TYPE_VAR_STRING ||
+         sql_field->real_field_type() == MYSQL_TYPE_STRING ||
+         sql_field->real_field_type() == MYSQL_TYPE_SET ||
+         sql_field->real_field_type() == MYSQL_TYPE_TINY_BLOB ||
+         sql_field->real_field_type() == MYSQL_TYPE_MEDIUM_BLOB ||
+         sql_field->real_field_type() == MYSQL_TYPE_LONG_BLOB ||
+         sql_field->real_field_type() == MYSQL_TYPE_BLOB ||
+         sql_field->real_field_type() == MYSQL_TYPE_ENUM))
     {
       Item *item;
       if (!(item= sql_field->default_value->expr->
@@ -3288,8 +3288,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 
     if (sql_field->default_value &&
         sql_field->default_value->expr->basic_const_item() &&
-        (sql_field->sql_type == MYSQL_TYPE_SET ||
-         sql_field->sql_type == MYSQL_TYPE_ENUM))
+        (sql_field->real_field_type() == MYSQL_TYPE_SET ||
+         sql_field->real_field_type() == MYSQL_TYPE_ENUM))
     {
       StringBuffer<MAX_FIELD_WIDTH> str;
       String *def= sql_field->default_value->expr->val_str(&str);
@@ -3301,7 +3301,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       else
       {
         not_found= false;
-        if (sql_field->sql_type == MYSQL_TYPE_SET)
+        if (sql_field->real_field_type() == MYSQL_TYPE_SET)
         {
           char *not_used;
           uint not_used2;
@@ -3357,12 +3357,12 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
             If we are replacing a BIT field, revert the increment
             of total_uneven_bit_length that was done above.
           */
-          if (sql_field->sql_type == MYSQL_TYPE_BIT &&
+          if (sql_field->real_field_type() == MYSQL_TYPE_BIT &&
               file->ha_table_flags() & HA_CAN_BIT_FIELD)
             total_uneven_bit_length-= sql_field->length & 7;
 
 	  sql_field->default_value=	dup_field->default_value;
-	  sql_field->sql_type=		dup_field->sql_type;
+	  sql_field->set_handler(dup_field->type_handler());
 
           /*
             If we are replacing a field with a BIT field, we need
@@ -3370,7 +3370,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
             increment total_uneven_bit_length here as this dup_field
             has already been processed.
           */
-          if (sql_field->sql_type == MYSQL_TYPE_BIT)
+          if (sql_field->real_field_type() == MYSQL_TYPE_BIT)
           {
             sql_field->pack_flag= FIELDFLAG_NUMBER;
             if (!(file->ha_table_flags() & HA_CAN_BIT_FIELD))
@@ -3404,7 +3404,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     }
     /* Don't pack rows in old tables if the user has requested this */
     if ((sql_field->flags & BLOB_FLAG) ||
-	(sql_field->sql_type == MYSQL_TYPE_VARCHAR &&
+	(sql_field->real_field_type() == MYSQL_TYPE_VARCHAR &&
          create_info->row_type != ROW_TYPE_FIXED))
       (*db_options)|= HA_OPTION_PACK_RECORD;
     it2.rewind();
@@ -3421,7 +3421,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 
     if (sql_field->prepare_create_field(&blob_columns, file->ha_table_flags()))
       DBUG_RETURN(TRUE);
-    if (sql_field->sql_type == MYSQL_TYPE_VARCHAR)
+    if (sql_field->real_field_type() == MYSQL_TYPE_VARCHAR)
       create_info->varchar= TRUE;
     sql_field->offset= record_offset;
     if (MTYP_TYPENR(sql_field->unireg_check) == Field::NEXT_NUMBER)
@@ -3721,8 +3721,8 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       cols2.rewind();
       if (key->type == Key::FULLTEXT)
       {
-	if ((sql_field->sql_type != MYSQL_TYPE_STRING &&
-	     sql_field->sql_type != MYSQL_TYPE_VARCHAR &&
+	if ((sql_field->real_field_type() != MYSQL_TYPE_STRING &&
+	     sql_field->real_field_type() != MYSQL_TYPE_VARCHAR &&
 	     !f_is_blob(sql_field->pack_flag)) ||
 	    sql_field->charset == &my_charset_bin ||
 	    sql_field->charset->mbminlen > 1 || // ucs2 doesn't work yet
@@ -3848,7 +3848,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 	if (f_is_blob(sql_field->pack_flag))
 	{
 	  key_part_length= MY_MIN(column->length,
-                               blob_length_by_type(sql_field->sql_type)
+                               blob_length_by_type(sql_field->real_field_type())
                                * sql_field->charset->mbmaxlen);
 	  if (key_part_length > max_key_length ||
 	      key_part_length > file->max_key_part_length())
@@ -3878,7 +3878,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
                  // is prefix length bigger than field length? 
                  (column->length > key_part_length ||
                   // can the field have a partial key? 
-                  !Field::type_can_have_key_part (sql_field->sql_type) ||
+                  !sql_field->type_handler()->type_can_have_key_part() ||
                   // a packed field can't be used in a partial key
                   f_is_packed(sql_field->pack_flag) ||
                   // does the storage engine allow prefixed search?
@@ -3922,12 +3922,12 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       if (!((*db_options) & HA_OPTION_NO_PACK_KEYS) &&
           !((create_info->table_options & HA_OPTION_NO_PACK_KEYS)) &&
 	  (key_part_length >= KEY_DEFAULT_PACK_LENGTH &&
-	   (sql_field->sql_type == MYSQL_TYPE_STRING ||
-	    sql_field->sql_type == MYSQL_TYPE_VARCHAR ||
+	   (sql_field->real_field_type() == MYSQL_TYPE_STRING ||
+	    sql_field->real_field_type() == MYSQL_TYPE_VARCHAR ||
 	    sql_field->pack_flag & FIELDFLAG_BLOB)))
       {
 	if ((column_nr == 0 && (sql_field->pack_flag & FIELDFLAG_BLOB)) ||
-            sql_field->sql_type == MYSQL_TYPE_VARCHAR)
+            sql_field->real_field_type() == MYSQL_TYPE_VARCHAR)
 	  key_info->flags|= HA_BINARY_PACK_KEY | HA_VAR_LENGTH_KEY;
 	else
 	  key_info->flags|= HA_PACK_KEY;
@@ -4028,7 +4028,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
     if (!sql_field->default_value &&
         !sql_field->has_default_function() &&
         (sql_field->flags & NOT_NULL_FLAG) &&
-        !is_timestamp_type(sql_field->sql_type))
+        !sql_field->is_timestamp_type())
     {
       sql_field->flags|= NO_DEFAULT_VALUE_FLAG;
       sql_field->pack_flag|= FIELDFLAG_NO_DEFAULT;
@@ -4036,7 +4036,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 
     if (thd->variables.sql_mode & MODE_NO_ZERO_DATE &&
         !sql_field->default_value && !sql_field->vcol_info &&
-        is_timestamp_type(sql_field->sql_type) &&
+        sql_field->is_timestamp_type() &&
         (sql_field->flags & NOT_NULL_FLAG) &&
         (type == Field::NONE || type == Field::TIMESTAMP_UN_FIELD))
     {
@@ -4219,7 +4219,7 @@ bool Column_definition::prepare_blob_field(THD *thd)
                static_cast<ulong>(MAX_FIELD_VARCHARLENGTH / charset->mbmaxlen));
       DBUG_RETURN(1);
     }
-    sql_type= MYSQL_TYPE_BLOB;
+    set_handler(&type_handler_blob);
     flags|= BLOB_FLAG;
     my_snprintf(warn_buff, sizeof(warn_buff), ER_THD(thd, ER_AUTO_CONVERT),
                 field_name.str,
@@ -4231,13 +4231,13 @@ bool Column_definition::prepare_blob_field(THD *thd)
 
   if ((flags & BLOB_FLAG) && length)
   {
-    if (sql_type == FIELD_TYPE_BLOB ||
-        sql_type == FIELD_TYPE_TINY_BLOB ||
-        sql_type == FIELD_TYPE_MEDIUM_BLOB)
+    if (real_field_type() == FIELD_TYPE_BLOB ||
+        real_field_type() == FIELD_TYPE_TINY_BLOB ||
+        real_field_type() == FIELD_TYPE_MEDIUM_BLOB)
     {
       /* The user has given a length to the blob column */
-      sql_type= get_blob_type_from_length(length);
-      pack_length= calc_pack_length(sql_type, 0);
+      set_handler(Type_handler::blob_type_handler(length));
+      pack_length= calc_pack_length(real_field_type(), 0);
     }
     length= 0;
   }
@@ -4262,7 +4262,8 @@ bool Column_definition::prepare_blob_field(THD *thd)
 
 bool Column_definition::sp_prepare_create_field(THD *thd, MEM_ROOT *mem_root)
 {
-  if (sql_type == MYSQL_TYPE_SET || sql_type == MYSQL_TYPE_ENUM)
+  if (real_field_type() == MYSQL_TYPE_SET ||
+      real_field_type() == MYSQL_TYPE_ENUM)
   {
     /*
       Pass "false" as the last argument to allocate TYPELIB values on mem_root,
@@ -4272,7 +4273,7 @@ bool Column_definition::sp_prepare_create_field(THD *thd, MEM_ROOT *mem_root)
       return true; // E.g. wrong values with commas: SET('a,b')
   }
 
-  if (sql_type == MYSQL_TYPE_BIT)
+  if (real_field_type() == MYSQL_TYPE_BIT)
     pack_flag= FIELDFLAG_NUMBER | FIELDFLAG_TREAT_BIT_AS_CHAR;
   create_length_to_internal_length();
   DBUG_ASSERT(default_value == 0);
@@ -6819,7 +6820,7 @@ bool mysql_compare_tables(TABLE *table,
     if (create_info->row_type == ROW_TYPE_DYNAMIC ||
         create_info->row_type == ROW_TYPE_PAGE ||
 	(tmp_new_field->flags & BLOB_FLAG) ||
-	(tmp_new_field->sql_type == MYSQL_TYPE_VARCHAR &&
+	(tmp_new_field->real_field_type() == MYSQL_TYPE_VARCHAR &&
 	create_info->row_type != ROW_TYPE_FIXED))
       create_info->table_options|= HA_OPTION_PACK_RECORD;
 
@@ -7611,10 +7612,10 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       If the '0000-00-00' value isn't allowed then raise the error_if_not_empty
       flag to allow ALTER TABLE only if the table to be altered is empty.
     */
-    if ((def->sql_type == MYSQL_TYPE_DATE ||
-         def->sql_type == MYSQL_TYPE_NEWDATE ||
-         def->sql_type == MYSQL_TYPE_DATETIME ||
-         def->sql_type == MYSQL_TYPE_DATETIME2) &&
+    if ((def->real_field_type() == MYSQL_TYPE_DATE ||
+         def->real_field_type() == MYSQL_TYPE_NEWDATE ||
+         def->real_field_type() == MYSQL_TYPE_DATETIME ||
+         def->real_field_type() == MYSQL_TYPE_DATETIME2) &&
          !alter_ctx->datetime_field &&
          !(~def->flags & (NO_DEFAULT_VALUE_FLAG | NOT_NULL_FLAG)) &&
          thd->variables.sql_mode & MODE_NO_ZERO_DATE)
@@ -7776,16 +7777,17 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
           - data type maximum length is 255.
           - key_part_length is 1016 (=254*4, where 4 is mbmaxlen)
          */
-        if (!Field::type_can_have_key_part(cfield->field->type()) ||
-            !Field::type_can_have_key_part(cfield->sql_type) ||
+        if (!cfield->field->type_handler()->type_can_have_key_part() ||
+            !cfield->type_handler()->type_can_have_key_part() ||
             /* spatial keys can't have sub-key length */
             (key_info->flags & HA_SPATIAL) ||
             (cfield->field->field_length == key_part_length &&
              !f_is_blob(key_part->key_type)) ||
-            (cfield->length && (((cfield->sql_type >= MYSQL_TYPE_TINY_BLOB &&
-                                  cfield->sql_type <= MYSQL_TYPE_BLOB) ? 
-                                blob_length_by_type(cfield->sql_type) :
-                                cfield->length) <
+            (cfield->length &&
+             (((cfield->real_field_type() >= MYSQL_TYPE_TINY_BLOB &&
+                cfield->real_field_type() <= MYSQL_TYPE_BLOB) ?
+                blob_length_by_type(cfield->real_field_type()) :
+                cfield->length) <
 	     key_part_length / key_part->field->charset()->mbmaxlen)))
 	  key_part_length= 0;			// Use whole field
       }
@@ -9494,7 +9496,7 @@ err_new_table_cleanup:
   {
     const char *f_val= 0;
     enum enum_mysql_timestamp_type t_type= MYSQL_TIMESTAMP_DATE;
-    switch (alter_ctx.datetime_field->sql_type)
+    switch (alter_ctx.datetime_field->real_field_type())
     {
       case MYSQL_TYPE_DATE:
       case MYSQL_TYPE_NEWDATE:
diff --git a/sql/sql_type.h b/sql/sql_type.h
index 4c507671f47..b94d73e8d23 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -584,6 +584,20 @@ public:
   {
     return MYSQL_TIMESTAMP_ERROR;
   }
+  virtual bool is_timestamp_type() const
+  {
+    return false;
+  }
+  /**
+    Check whether a field type can be partially indexed by a key.
+    @param  type   field type
+    @retval true   Type can have a prefixed key
+    @retval false  Type can not have a prefixed key
+  */
+  virtual bool type_can_have_key_part() const
+  {
+    return false;
+  }
   /**
     Prepared statement long data:
     Check whether this parameter data type is compatible with long data.
@@ -1837,6 +1851,10 @@ public:
   {
     return MYSQL_TIMESTAMP_DATETIME;
   }
+  bool is_timestamp_type() const
+  {
+    return true;
+  }
   uint Item_decimal_scale(const Item *item) const
   {
     return Item_decimal_scale_with_seconds(item);
@@ -1938,7 +1956,17 @@ public:
 };
 
 
-class Type_handler_string: public Type_handler_string_result
+class Type_handler_longstr: public Type_handler_string_result
+{
+public:
+  bool type_can_have_key_part() const
+  {
+    return true;
+  }
+};
+
+
+class Type_handler_string: public Type_handler_longstr
 {
   static const Name m_name_char;
 public:
@@ -1979,7 +2007,7 @@ public:
 };
 
 
-class Type_handler_varchar: public Type_handler_string_result
+class Type_handler_varchar: public Type_handler_longstr
 {
   static const Name m_name_varchar;
 public:
@@ -2004,7 +2032,7 @@ public:
 };
 
 
-class Type_handler_blob_common: public Type_handler_string_result
+class Type_handler_blob_common: public Type_handler_longstr
 {
 public:
   virtual ~Type_handler_blob_common() { }
@@ -2101,6 +2129,10 @@ public:
   enum_field_types field_type() const { return MYSQL_TYPE_GEOMETRY; }
   bool is_param_long_data_type() const { return true; }
   const Type_handler *type_handler_for_comparison() const;
+  bool type_can_have_key_part() const
+  {
+    return true;
+  }
   bool subquery_type_allows_materialization(const Item *inner,
                                             const Item *outer) const
   {
@@ -2213,6 +2245,10 @@ public:
   {
     return m_type_handler->mysql_timestamp_type();
   }
+  bool is_timestamp_type() const
+  {
+    return m_type_handler->is_timestamp_type();
+  }
   void set_handler(const Type_handler *other)
   {
     m_type_handler= other;
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 1fa8018ab7c..e8120c325fb 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -3817,7 +3817,7 @@ sp_proc_stmt_return:
 
             i= new (thd->mem_root)
                  sp_instr_freturn(sp->instructions(), lex->spcont, $3,
-                                  sp->m_return_field_def.sql_type, lex);
+                                  sp->m_return_field_def.type_handler(), lex);
             if (i == NULL || sp->add_instr(i))
               MYSQL_YYABORT;
             sp->m_flags|= sp_head::HAS_RETURN;
@@ -6116,7 +6116,7 @@ field_type_or_serial:
           field_def
         | SERIAL_SYM
           {
-            Lex->last_field->sql_type= MYSQL_TYPE_LONGLONG;
+            Lex->last_field->set_handler(&type_handler_longlong);
             Lex->last_field->flags|= AUTO_INCREMENT_FLAG | NOT_NULL_FLAG
                                      | UNSIGNED_FLAG | UNIQUE_KEY_FLAG;
           }
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 1cbb69d2c48..91b541410bc 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -3392,7 +3392,7 @@ sp_proc_stmt_return:
 
             i= new (thd->mem_root)
                  sp_instr_freturn(sp->instructions(), lex->spcont, $3,
-                                  sp->m_return_field_def.sql_type, lex);
+                                  sp->m_return_field_def.type_handler(), lex);
             if (i == NULL || sp->add_instr(i))
               MYSQL_YYABORT;
             sp->m_flags|= sp_head::HAS_RETURN;
@@ -5988,7 +5988,7 @@ field_type_or_serial:
           field_def
         | SERIAL_SYM
           {
-            Lex->last_field->sql_type= MYSQL_TYPE_LONGLONG;
+            Lex->last_field->set_handler(&type_handler_longlong);
             Lex->last_field->flags|= AUTO_INCREMENT_FLAG | NOT_NULL_FLAG
                                      | UNSIGNED_FLAG | UNIQUE_KEY_FLAG;
           }
diff --git a/sql/table.cc b/sql/table.cc
index b57d45b0f1f..eac12578b15 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1753,6 +1753,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     Virtual_column_info *vcol_info= 0;
     uint gis_length, gis_decimals, srid= 0;
     Field::utype unireg_check;
+    const Type_handler *handler;
 
     if (new_frm_ver >= 3)
     {
@@ -1964,9 +1965,11 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
     unireg_check= (Field::utype) MTYP_TYPENR(unireg_type);
     name.str= fieldnames.type_names[i];
     name.length= strlen(name.str);
+    if (!(handler= Type_handler::get_handler_by_real_type(field_type)))
+      goto err; // Not supported field type
     *field_ptr= reg_field=
       make_field(share, &share->mem_root, record+recpos, (uint32) field_length,
-		 null_pos, null_bit_pos, pack_flag, field_type, charset,
+		 null_pos, null_bit_pos, pack_flag, handler, charset,
 		 geom_type, srid, unireg_check,
 		 (interval_nr ? share->intervals+interval_nr-1 : NULL),
 		 &name);
diff --git a/sql/unireg.cc b/sql/unireg.cc
index 268ee24f509..655cfa20c61 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -626,7 +626,7 @@ static bool pack_header(THD *thd, uchar *forminfo,
       We mark first TIMESTAMP field with NOW() in DEFAULT or ON UPDATE
       as auto-update field.
     */
-    if (field->sql_type == MYSQL_TYPE_TIMESTAMP &&
+    if (field->real_field_type() == MYSQL_TYPE_TIMESTAMP &&
         MTYP_TYPENR(field->unireg_check) != Field::NONE &&
 	!time_stamp_pos)
       time_stamp_pos= (uint) field->offset+ (uint) data_offset + 1;
@@ -808,8 +808,8 @@ static bool pack_fields(uchar **buff_arg, List<Create_field> &create_fields,
     int2store(buff+8,field->pack_flag);
     buff[10]= (uchar) field->unireg_check;
     buff[12]= (uchar) field->interval_id;
-    buff[13]= (uchar) field->sql_type;
-    if (field->sql_type == MYSQL_TYPE_GEOMETRY)
+    buff[13]= (uchar) field->real_field_type();
+    if (field->real_field_type() == MYSQL_TYPE_GEOMETRY)
     {
       buff[11]= 0;
       buff[14]= (uchar) field->geom_type;
@@ -954,7 +954,7 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
                                 null_pos + null_count / 8,
                                 null_count & 7,
                                 field->pack_flag,
-                                field->sql_type,
+                                field->type_handler(),
                                 field->charset,
                                 field->geom_type, field->srid,
                                 field->unireg_check,
@@ -976,7 +976,8 @@ static bool make_empty_rec(THD *thd, uchar *buff, uint table_options,
       null_count++;
     }
 
-    if (field->sql_type == MYSQL_TYPE_BIT && !f_bit_as_char(field->pack_flag))
+    if (field->real_field_type() == MYSQL_TYPE_BIT &&
+        !f_bit_as_char(field->pack_flag))
       null_count+= field->length & 7;
 
     if (field->default_value && !field->default_value->flags &&
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index dcfe72b0531..f9791f5fa97 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -735,7 +735,7 @@ ha_innobase::check_if_supported_inplace_alter(
 			is TIMESTAMP and it is defined as NOT NULL and
 			it has either constant default or function default
 			we must use "Copy" method. */
-			if (is_timestamp_type(def->sql_type)) {
+			if (def->is_timestamp_type()) {
 				if ((def->flags & NOT_NULL_FLAG) != 0 && // NOT NULL
 					(def->default_value != NULL || // constant default ?
 					 def->unireg_check != Field::NONE)) { // function default
diff --git a/storage/perfschema/pfs_engine_table.cc b/storage/perfschema/pfs_engine_table.cc
index 5f37d43acf1..f3c6edcef59 100644
--- a/storage/perfschema/pfs_engine_table.cc
+++ b/storage/perfschema/pfs_engine_table.cc
@@ -396,7 +396,7 @@ void PFS_engine_table::set_field_enum(Field *f, ulonglong value)
 
 void PFS_engine_table::set_field_timestamp(Field *f, ulonglong value)
 {
-  DBUG_ASSERT(is_timestamp_type(f->real_type()));
+  DBUG_ASSERT(f->type_handler()->is_timestamp_type());
   Field_timestamp *f2= (Field_timestamp*) f;
   f2->store_TIME((long)(value / 1000000), (value % 1000000));
 }
diff --git a/storage/xtradb/handler/handler0alter.cc b/storage/xtradb/handler/handler0alter.cc
index b73ed019c6f..39e7c952f5d 100644
--- a/storage/xtradb/handler/handler0alter.cc
+++ b/storage/xtradb/handler/handler0alter.cc
@@ -452,7 +452,7 @@ ha_innobase::check_if_supported_inplace_alter(
 			is TIMESTAMP and it is defined as NOT NULL and
 			it has either constant default or function default
 			we must use "Copy" method. */
-			if (is_timestamp_type(def->sql_type)) {
+			if (def->is_timestamp_type()) {
 				if ((def->flags & NOT_NULL_FLAG) != 0 && // NOT NULL
 					(def->default_value != NULL || // constant default ?
 					 def->unireg_check != Field::NONE)) { // function default