Adding support for MySQL-5.6 temporal column types:

  TIME, DATETIME, TIMESTAMP


added:
  mysql-test/r/type_temporal_mysql56.result
  mysql-test/std_data/mysql56datetime.MYD
  mysql-test/std_data/mysql56datetime.MYI
  mysql-test/std_data/mysql56datetime.frm
  mysql-test/std_data/mysql56time.MYD
  mysql-test/std_data/mysql56time.MYI
  mysql-test/std_data/mysql56time.frm
  mysql-test/std_data/mysql56timestamp.MYD
  mysql-test/std_data/mysql56timestamp.MYI
  mysql-test/std_data/mysql56timestamp.frm
  mysql-test/suite/rpl/r/rpl_temporal_mysql56.result
  mysql-test/suite/rpl/t/rpl_temporal_mysql56.test
  mysql-test/t/type_temporal_mysql56.test
  sql/compat56.cc
  sql/compat56.h
modified:
  client/mysqlbinlog.cc
  include/my_time.h
  include/mysql.h.pp
  include/mysql_com.h
  mysql-test/r/statistics.result
  mysql-test/r/strict.result
  mysql-test/suite/binlog/r/binlog_mysqlbinlog_row_innodb.result
  mysql-test/suite/binlog/r/binlog_mysqlbinlog_row_myisam.result
  sql-common/my_time.c
  sql/CMakeLists.txt
  sql/field.cc
  sql/field.h
  sql/item.cc
  sql/item_strfunc.cc
  sql/item_sum.cc
  sql/item_timefunc.cc
  sql/log_event.cc
  sql/opt_range.cc
  sql/opt_table_elimination.cc
  sql/protocol.cc
  sql/rpl_utility.cc
  sql/rpl_utility.h
  sql/sql_partition.cc
  sql/sql_prepare.cc
  sql/sql_select.cc
  sql/sql_table.cc
  sql/table.cc
  storage/perfschema/pfs_engine_table.cc

20 files changed, 1457 insertions, 526 deletions

diff --git a/sql/CMakeLists.txt b/sql/CMakeLists.txt
index bf1d0bfdc9a..dc6ba20811d 100644
--- a/sql/CMakeLists.txt
+++ b/sql/CMakeLists.txt
@@ -37,7 +37,7 @@ IF(SSL_DEFINES)
 ENDIF()
 
 SET (SQL_SOURCE
-              ../sql-common/client.c derror.cc des_key_file.cc
+              ../sql-common/client.c compat56.cc derror.cc des_key_file.cc
                discover.cc ../libmysql/errmsg.c field.cc  field_conv.cc 
                filesort_utils.cc
                filesort.cc gstream.cc sha2.cc
diff --git a/sql/compat56.cc b/sql/compat56.cc
new file mode 100644
index 00000000000..3bd6b21a154
--- /dev/null
+++ b/sql/compat56.cc
@@ -0,0 +1,445 @@
+/*
+   Copyright (c) 2004, 2012, Oracle and/or its affiliates.
+   Copyright (c) 2013, MariaDB Foundation.
+
+ This program is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; version 2 of the License.
+
+ This program is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program; if not, write to the Free Software
+ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+#include "my_global.h"
+#include "compat56.h"
+#include "myisampack.h"
+#include "my_time.h"
+
+/*** MySQL56 TIME low-level memory and disk representation routines ***/
+
+/*
+  In-memory format:
+
+   1  bit sign          (Used for sign, when on disk)
+   1  bit unused        (Reserved for wider hour range, e.g. for intervals)
+   10 bit hour          (0-836)
+   6  bit minute        (0-59)
+   6  bit second        (0-59)
+  24  bits microseconds (0-999999)
+
+ Total: 48 bits = 6 bytes
+   Suhhhhhh.hhhhmmmm.mmssssss.ffffffff.ffffffff.ffffffff
+*/
+
+
+/**
+  Convert time value to MySQL56 numeric packed representation.
+  
+  @param    ltime   The value to convert.
+  @return           Numeric packed representation.
+*/
+longlong TIME_to_longlong_time_packed(const MYSQL_TIME *ltime)
+{
+  /* If month is 0, we mix day with hours: "1 00:10:10" -> "24:00:10" */
+  long hms= (((ltime->month ? 0 : ltime->day * 24) + ltime->hour) << 12) |
+            (ltime->minute << 6) | ltime->second;
+  longlong tmp= MY_PACKED_TIME_MAKE(hms, ltime->second_part);
+  return ltime->neg ? -tmp : tmp;
+}
+
+
+
+/**
+  Convert MySQL56 time packed numeric representation to time.
+
+  @param  OUT ltime  The MYSQL_TIME variable to set.
+  @param      tmp    The packed numeric representation.
+*/
+void TIME_from_longlong_time_packed(MYSQL_TIME *ltime, longlong tmp)
+{
+  long hms;
+  if ((ltime->neg= (tmp < 0)))
+    tmp= -tmp;
+  hms= MY_PACKED_TIME_GET_INT_PART(tmp);
+  ltime->year=   (uint) 0;
+  ltime->month=  (uint) 0;
+  ltime->day=    (uint) 0;
+  ltime->hour=   (uint) (hms >> 12) % (1 << 10); /* 10 bits starting at 12th */
+  ltime->minute= (uint) (hms >> 6)  % (1 << 6);  /* 6 bits starting at 6th   */
+  ltime->second= (uint)  hms        % (1 << 6);  /* 6 bits starting at 0th   */
+  ltime->second_part= MY_PACKED_TIME_GET_FRAC_PART(tmp);
+  ltime->time_type= MYSQL_TIMESTAMP_TIME;
+}
+
+
+/**
+  Calculate binary size of MySQL56 packed numeric time representation.
+  
+  @param   dec   Precision.
+*/
+uint my_time_binary_length(uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  return 3 + (dec + 1) / 2;
+}
+
+
+/*
+  On disk we convert from signed representation to unsigned
+  representation using TIMEF_OFS, so all values become binary comparable.
+*/
+#define TIMEF_OFS 0x800000000000LL
+#define TIMEF_INT_OFS 0x800000LL
+
+
+/**
+  Convert MySQL56 in-memory numeric time representation to on-disk representation
+  
+  @param       nr   Value in packed numeric time format.
+  @param   OUT ptr  The buffer to put value at.
+  @param       dec  Precision.
+*/
+void my_time_packed_to_binary(longlong nr, uchar *ptr, uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  /* Make sure the stored value was previously properly rounded or truncated */
+  DBUG_ASSERT((MY_PACKED_TIME_GET_FRAC_PART(nr) % 
+              (int) log_10_int[TIME_SECOND_PART_DIGITS - dec]) == 0);
+
+  switch (dec)
+  {
+  case 0:
+  default:
+    mi_int3store(ptr, TIMEF_INT_OFS + MY_PACKED_TIME_GET_INT_PART(nr));
+    break;
+
+  case 1:
+  case 2:
+    mi_int3store(ptr, TIMEF_INT_OFS + MY_PACKED_TIME_GET_INT_PART(nr));
+    ptr[3]= (unsigned char) (char) (MY_PACKED_TIME_GET_FRAC_PART(nr) / 10000);
+    break;
+
+  case 4:
+  case 3:
+    mi_int3store(ptr, TIMEF_INT_OFS + MY_PACKED_TIME_GET_INT_PART(nr));
+    mi_int2store(ptr + 3, MY_PACKED_TIME_GET_FRAC_PART(nr) / 100);
+    break;
+
+  case 5:
+  case 6:
+    mi_int6store(ptr, nr + TIMEF_OFS);
+    break;
+  }
+}
+
+
+/**
+  Convert MySQL56 on-disk time representation to in-memory packed numeric 
+  representation.
+  
+  @param   ptr  The pointer to read the value at.
+  @param   dec  Precision.
+  @return       Packed numeric time representation.
+*/
+longlong my_time_packed_from_binary(const uchar *ptr, uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+
+  switch (dec)
+  {
+  case 0:
+  default:
+    {
+      longlong intpart= mi_uint3korr(ptr) - TIMEF_INT_OFS;
+      return MY_PACKED_TIME_MAKE_INT(intpart);
+    }
+  case 1:
+  case 2:
+    {
+      longlong intpart= mi_uint3korr(ptr) - TIMEF_INT_OFS;
+      int frac= (uint) ptr[3];
+      if (intpart < 0 && frac)
+      {
+        /*
+          Negative values are stored with reverse fractional part order,
+          for binary sort compatibility.
+
+            Disk value  intpart frac   Time value   Memory value
+            800000.00    0      0      00:00:00.00  0000000000.000000
+            7FFFFF.FF   -1      255   -00:00:00.01  FFFFFFFFFF.FFD8F0
+            7FFFFF.9D   -1      99    -00:00:00.99  FFFFFFFFFF.F0E4D0
+            7FFFFF.00   -1      0     -00:00:01.00  FFFFFFFFFF.000000
+            7FFFFE.FF   -1      255   -00:00:01.01  FFFFFFFFFE.FFD8F0
+            7FFFFE.F6   -2      246   -00:00:01.10  FFFFFFFFFE.FE7960
+
+            Formula to convert fractional part from disk format
+            (now stored in "frac" variable) to absolute value: "0x100 - frac".
+            To reconstruct in-memory value, we shift
+            to the next integer value and then substruct fractional part.
+        */
+        intpart++;    /* Shift to the next integer value */
+        frac-= 0x100; /* -(0x100 - frac) */
+      }
+      return MY_PACKED_TIME_MAKE(intpart, frac * 10000);
+    }
+
+  case 3:
+  case 4:
+    {
+      longlong intpart= mi_uint3korr(ptr) - TIMEF_INT_OFS;
+      int frac= mi_uint2korr(ptr + 3);
+      if (intpart < 0 && frac)
+      {
+        /*
+          Fix reverse fractional part order: "0x10000 - frac".
+          See comments for FSP=1 and FSP=2 above.
+        */
+        intpart++;      /* Shift to the next integer value */
+        frac-= 0x10000; /* -(0x10000-frac) */
+      }
+      return MY_PACKED_TIME_MAKE(intpart, frac * 100);
+    }
+
+  case 5:
+  case 6:
+    return ((longlong) mi_uint6korr(ptr)) - TIMEF_OFS;
+  }
+}
+
+
+/*** MySQL56 DATETIME low-level memory and disk representation routines ***/
+
+/*
+    1 bit  sign            (used when on disk)
+   17 bits year*13+month   (year 0-9999, month 0-12)
+    5 bits day             (0-31)
+    5 bits hour            (0-23)
+    6 bits minute          (0-59)
+    6 bits second          (0-59)
+   24 bits microseconds    (0-999999)
+
+   Total: 64 bits = 8 bytes
+
+   SYYYYYYY.YYYYYYYY.YYdddddh.hhhhmmmm.mmssssss.ffffffff.ffffffff.ffffffff
+*/
+
+/**
+  Convert datetime to MySQL56 packed numeric datetime representation.
+  @param ltime  The value to convert.
+  @return       Packed numeric representation of ltime.
+*/
+longlong TIME_to_longlong_datetime_packed(const MYSQL_TIME *ltime)
+{
+  longlong ymd= ((ltime->year * 13 + ltime->month) << 5) | ltime->day;
+  longlong hms= (ltime->hour << 12) | (ltime->minute << 6) | ltime->second;
+  longlong tmp= MY_PACKED_TIME_MAKE(((ymd << 17) | hms), ltime->second_part);
+  DBUG_ASSERT(!check_datetime_range(ltime)); /* Make sure no overflow */
+  return ltime->neg ? -tmp : tmp;
+}
+
+
+/**
+  Convert MySQL56 packed numeric datetime representation to MYSQL_TIME.
+  @param OUT  ltime The datetime variable to convert to.
+  @param      tmp   The packed numeric datetime value.
+*/
+void TIME_from_longlong_datetime_packed(MYSQL_TIME *ltime, longlong tmp)
+{
+  longlong ymd, hms;
+  longlong ymdhms, ym;
+  if ((ltime->neg= (tmp < 0)))
+    tmp= -tmp;
+
+  ltime->second_part= MY_PACKED_TIME_GET_FRAC_PART(tmp);
+  ymdhms= MY_PACKED_TIME_GET_INT_PART(tmp);
+
+  ymd= ymdhms >> 17;
+  ym= ymd >> 5;
+  hms= ymdhms % (1 << 17);
+
+  ltime->day= ymd % (1 << 5);
+  ltime->month= ym % 13;
+  ltime->year= ym / 13;
+
+  ltime->second= hms % (1 << 6);
+  ltime->minute= (hms >> 6) % (1 << 6);
+  ltime->hour= (hms >> 12);
+  
+  ltime->time_type= MYSQL_TIMESTAMP_DATETIME;
+}
+
+
+/**
+  Calculate binary size of MySQL56 packed datetime representation.
+  @param dec  Precision.
+*/
+uint my_datetime_binary_length(uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  return 5 + (dec + 1) / 2;
+}
+
+
+/*
+  On disk we store as unsigned number with DATETIMEF_INT_OFS offset,
+  for HA_KETYPE_BINARY compatibilty purposes.
+*/
+#define DATETIMEF_INT_OFS 0x8000000000LL
+
+
+/**
+  Convert MySQL56 on-disk datetime representation
+  to in-memory packed numeric representation.
+
+  @param ptr   The pointer to read value at.
+  @param dec   Precision.
+  @return      In-memory packed numeric datetime representation.
+*/
+longlong my_datetime_packed_from_binary(const uchar *ptr, uint dec)
+{
+  longlong intpart= mi_uint5korr(ptr) - DATETIMEF_INT_OFS;
+  int frac;
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  switch (dec)
+  {
+  case 0:
+  default:
+    return MY_PACKED_TIME_MAKE_INT(intpart);
+  case 1:
+  case 2:
+    frac= ((int) (signed char) ptr[5]) * 10000;
+    break;
+  case 3:
+  case 4:
+    frac= mi_sint2korr(ptr + 5) * 100;
+    break;
+  case 5:
+  case 6:
+    frac= mi_sint3korr(ptr + 5);
+    break;
+  }
+  return MY_PACKED_TIME_MAKE(intpart, frac);
+}
+
+
+/**
+  Store MySQL56 in-memory numeric packed datetime representation to disk.
+
+  @param      nr  In-memory numeric packed datetime representation.
+  @param OUT  ptr The pointer to store at.
+  @param      dec Precision, 1-6.
+*/
+void my_datetime_packed_to_binary(longlong nr, uchar *ptr, uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  /* The value being stored must have been properly rounded or truncated */
+  DBUG_ASSERT((MY_PACKED_TIME_GET_FRAC_PART(nr) %
+              (int) log_10_int[TIME_SECOND_PART_DIGITS - dec]) == 0);
+
+  mi_int5store(ptr, MY_PACKED_TIME_GET_INT_PART(nr) + DATETIMEF_INT_OFS);
+  switch (dec)
+  {
+  case 0:
+  default:
+    break;
+  case 1:
+  case 2:
+    ptr[5]= (unsigned char) (char) (MY_PACKED_TIME_GET_FRAC_PART(nr) / 10000);
+    break;
+  case 3:
+  case 4:
+    mi_int2store(ptr + 5, MY_PACKED_TIME_GET_FRAC_PART(nr) / 100);
+    break;
+  case 5:
+  case 6:
+    mi_int3store(ptr + 5, MY_PACKED_TIME_GET_FRAC_PART(nr));
+  }
+}
+
+
+/*** MySQL56 TIMESTAMP low-level memory and disk representation routines ***/
+
+/**
+  Calculate on-disk size of a timestamp value.
+
+  @param  dec  Precision.
+*/
+uint my_timestamp_binary_length(uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  return 4 + (dec + 1) / 2;
+}
+
+
+/**
+  Convert MySQL56 binary timestamp representation to in-memory representation.
+
+  @param  OUT tm  The variable to convert to.
+  @param      ptr The pointer to read the value from.
+  @param      dec Precision.
+*/
+void my_timestamp_from_binary(struct timeval *tm, const uchar *ptr, uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  tm->tv_sec= mi_uint4korr(ptr);
+  switch (dec)
+  {
+    case 0:
+    default:
+      tm->tv_usec= 0;
+      break;
+    case 1:
+    case 2:
+      tm->tv_usec= ((int) ptr[4]) * 10000;
+      break;
+    case 3:
+    case 4:
+      tm->tv_usec= mi_sint2korr(ptr + 4) * 100;
+      break;
+    case 5:
+    case 6:
+      tm->tv_usec= mi_sint3korr(ptr + 4);
+  }
+}
+
+
+/**
+  Convert MySQL56 in-memory timestamp representation to on-disk representation.
+
+  @param        tm   The value to convert.
+  @param  OUT   ptr  The pointer to store the value to.
+  @param        dec  Precision.
+*/
+void my_timestamp_to_binary(const struct timeval *tm, uchar *ptr, uint dec)
+{
+  DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  /* Stored value must have been previously properly rounded or truncated */
+  DBUG_ASSERT((tm->tv_usec %
+               (int) log_10_int[TIME_SECOND_PART_DIGITS - dec]) == 0);
+  mi_int4store(ptr, tm->tv_sec);
+  switch (dec)
+  {
+    case 0:
+    default:
+      break;
+    case 1:
+    case 2:
+      ptr[4]= (unsigned char) (char) (tm->tv_usec / 10000);
+      break;
+    case 3:
+    case 4:
+      mi_int2store(ptr + 4, tm->tv_usec / 100);
+      break;
+      /* Impossible second precision. Fall through */
+    case 5:
+    case 6:
+      mi_int3store(ptr + 4, tm->tv_usec);
+  }
+}
+
+/****************************************/
diff --git a/sql/compat56.h b/sql/compat56.h
new file mode 100644
index 00000000000..bb5e2670f7d
--- /dev/null
+++ b/sql/compat56.h
@@ -0,0 +1,46 @@
+#ifndef COMPAT56_H_INCLUDED
+#define COMPAT56_H_INCLUDED
+/*
+   Copyright (c) 2004, 2012, Oracle and/or its affiliates.
+   Copyright (c) 2013  MariaDB Foundation.
+
+ This program is free software; you can redistribute it and/or modify
+ it under the terms of the GNU General Public License as published by
+ the Free Software Foundation; version 2 of the License.
+
+ This program is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program; if not, write to the Free Software
+ Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA */
+
+
+/** MySQL56 routines and macros **/
+#define MY_PACKED_TIME_GET_INT_PART(x)     ((x) >> 24)
+#define MY_PACKED_TIME_GET_FRAC_PART(x)    ((x) % (1LL << 24))
+#define MY_PACKED_TIME_MAKE(i, f)          ((((longlong) (i)) << 24) + (f))
+#define MY_PACKED_TIME_MAKE_INT(i)         ((((longlong) (i)) << 24))
+
+longlong TIME_to_longlong_datetime_packed(const MYSQL_TIME *);
+longlong TIME_to_longlong_time_packed(const MYSQL_TIME *);
+
+void TIME_from_longlong_datetime_packed(MYSQL_TIME *ltime, longlong nr);
+void TIME_from_longlong_time_packed(MYSQL_TIME *ltime, longlong nr);
+
+void my_datetime_packed_to_binary(longlong nr, uchar *ptr, uint dec);
+longlong my_datetime_packed_from_binary(const uchar *ptr, uint dec);
+uint my_datetime_binary_length(uint dec);
+
+void my_time_packed_to_binary(longlong nr, uchar *ptr, uint dec);
+longlong my_time_packed_from_binary(const uchar *ptr, uint dec);
+uint my_time_binary_length(uint dec);
+
+void my_timestamp_to_binary(const struct timeval *tm, uchar *ptr, uint dec);
+void my_timestamp_from_binary(struct timeval *tm, const uchar *ptr, uint dec);
+uint my_timestamp_binary_length(uint dec);
+/** End of MySQL routines and macros **/
+
+#endif /* COMPAT56_H_INCLUDED */
diff --git a/sql/field.cc b/sql/field.cc
index e85903d76c6..c51a04ba140 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -87,6 +87,7 @@ const char field_separator=',';
 #define FIELDTYPE_NUM (FIELDTYPE_TEAR_FROM + (255 - FIELDTYPE_TEAR_TO))
 static inline int field_type2index (enum_field_types field_type)
 {
+  field_type= real_type_to_type(field_type);
   return (field_type < FIELDTYPE_TEAR_FROM ?
           field_type :
           ((int)FIELDTYPE_TEAR_FROM) + (field_type - FIELDTYPE_TEAR_TO) - 1);
@@ -947,8 +948,10 @@ static enum_field_types field_types_merge_rules [FIELDTYPE_NUM][FIELDTYPE_NUM]=
 enum_field_types Field::field_type_merge(enum_field_types a,
                                          enum_field_types b)
 {
-  DBUG_ASSERT(a < FIELDTYPE_TEAR_FROM || a > FIELDTYPE_TEAR_TO);
-  DBUG_ASSERT(b < FIELDTYPE_TEAR_FROM || b > FIELDTYPE_TEAR_TO);
+  DBUG_ASSERT(real_type_to_type(a) < FIELDTYPE_TEAR_FROM ||
+              real_type_to_type(a) > FIELDTYPE_TEAR_TO);
+  DBUG_ASSERT(real_type_to_type(b) < FIELDTYPE_TEAR_FROM ||
+              real_type_to_type(b) > FIELDTYPE_TEAR_TO);
   return field_types_merge_rules[field_type2index(a)]
                                 [field_type2index(b)];
 }
@@ -1042,8 +1045,8 @@ CPP_UNNAMED_NS_END
 
 Item_result Field::result_merge_type(enum_field_types field_type)
 {
-  DBUG_ASSERT(field_type < FIELDTYPE_TEAR_FROM || field_type
-              > FIELDTYPE_TEAR_TO);
+  DBUG_ASSERT(real_type_to_type(field_type) < FIELDTYPE_TEAR_FROM ||
+              real_type_to_type(field_type) > FIELDTYPE_TEAR_TO);
   return field_types_result_type[field_type2index(field_type)];
 }
 
@@ -1130,6 +1133,111 @@ void Field::make_sort_key(uchar *buff,uint length)
 
 
 /**
+  @brief
+  Determine the relative position of the field value in a numeric interval
+
+  @details
+  The function returns a double number between 0.0 and 1.0 as the relative
+  position of the value of the this field in the numeric interval of [min,max].
+  If the value is not in the interval the the function returns 0.0 when
+  the value is less than min, and, 1.0 when the value is greater than max.
+
+  @param  min  value of the left end of the interval
+  @param  max  value of the right end of the interval
+
+  @return
+  relative position of the field value in the numeric interval [min,max] 
+*/
+
+double Field::pos_in_interval_val_real(Field *min, Field *max)
+{
+  double n, d;
+  n= val_real() - min->val_real();
+  if (n < 0)
+    return 0.0;
+  d= max->val_real() - min->val_real();
+  if (d <= 0)
+    return 1.0;
+  return min(n/d, 1.0);
+}
+
+
+static
+inline ulonglong char_prefix_to_ulonglong(uchar *src)
+{
+  uint sz= sizeof(ulonglong);
+  for (uint i= 0; i < sz/2; i++)
+  {
+    uchar tmp= src[i];
+    src[i]= src[sz-1-i];
+    src[sz-1-i]= tmp;
+  }
+  return uint8korr(src); 
+}
+
+
+/**
+  @brief
+  Determine the relative position of the field value in a string interval
+
+  @details
+  The function returns a double number between 0.0 and 1.0 as the relative
+  position of the value of the this field in the string interval of [min,max].
+  If the value is not in the interval the the function returns 0.0 when
+  the value is less than min, and, 1.0 when the value is greater than max.
+
+  @note
+  To calculate the relative position of the string value v in the interval
+  [min, max] the function first converts the beginning of these three
+  strings v, min, max into the strings that are used for byte comparison.
+  For each string not more sizeof(ulonglong) first bytes are taken
+  from the result of conversion. Then these bytes are interpreted as the
+  big-endian representation of an ulonglong integer. The values of these
+  integer numbers obtained for the strings v, min, max are used to calculate
+  the position of v in [min,max] in the same way is it's done for numeric
+  fields (see Field::pos_in_interval_val_real).
+
+  @todo
+  Improve the procedure for the case when min and max have the same
+  beginning
+     
+  @param  min  value of the left end of the interval
+  @param  max  value of the right end of the interval
+
+  @return
+  relative position of the field value in the string interval [min,max] 
+*/
+
+double Field::pos_in_interval_val_str(Field *min, Field *max, uint data_offset)
+{
+  uchar mp_prefix[sizeof(ulonglong)];
+  uchar minp_prefix[sizeof(ulonglong)];
+  uchar maxp_prefix[sizeof(ulonglong)];
+  ulonglong mp, minp, maxp;
+  my_strnxfrm(charset(), mp_prefix, sizeof(mp),
+              ptr + data_offset,
+              data_length());
+  my_strnxfrm(charset(), minp_prefix, sizeof(minp),
+              min->ptr + data_offset,
+              min->data_length());
+  my_strnxfrm(charset(), maxp_prefix, sizeof(maxp),
+              max->ptr + data_offset,
+              max->data_length());
+  mp= char_prefix_to_ulonglong(mp_prefix);
+  minp= char_prefix_to_ulonglong(minp_prefix);
+  maxp= char_prefix_to_ulonglong(maxp_prefix);
+  double n, d;
+  n= mp - minp;
+  if (n < 0)
+    return 0.0;
+  d= maxp - minp;
+  if (d <= 0)
+    return 1.0;
+  return min(n/d, 1.0);
+}
+
+
+/**
   Numeric fields base class constructor.
 */
 Field_num::Field_num(uchar *ptr_arg,uint32 len_arg, uchar *null_ptr_arg,
@@ -1275,36 +1383,6 @@ out_of_range:
 
 
 /**
-  @brief
-  Determine the relative position of the field value in a numeric interval
-
-  @details
-  The function returns a double number between 0.0 and 1.0 as the relative
-  position of the value of the this field in the numeric interval of [min,max].
-  If the value is not in the interval the the function returns 0.0 when
-  the value is less than min, and, 1.0 when the value is greater than max.
-
-  @param  min  value of the left end of the interval
-  @param  max  value of the right end of the interval
-
-  @return
-  relative position of the field value in the numeric interval [min,max] 
-*/
-
-double Field_num::pos_in_interval(Field *min, Field *max)
-{
-  double n, d;
-  n= val_real() - min->val_real();
-  if (n < 0)
-    return 0.0;
-  d= max->val_real() - min->val_real();
-  if (d <= 0)
-    return 1.0;
-  return min(n/d, 1.0);
-}
-
-
-/**
   Process decimal library return codes and issue warnings for overflow and
   truncation.
 
@@ -4485,13 +4563,12 @@ Field_timestamp::Field_timestamp(uchar *ptr_arg, uint32 len_arg,
                                  uchar *null_ptr_arg, uchar null_bit_arg,
 				 enum utype unireg_check_arg,
 				 const char *field_name_arg,
-				 TABLE_SHARE *share,
-				 CHARSET_INFO *cs)
-  :Field_str(ptr_arg, len_arg, null_ptr_arg, null_bit_arg,
-	     unireg_check_arg, field_name_arg, cs)
+				 TABLE_SHARE *share)
+  :Field_temporal(ptr_arg, len_arg, null_ptr_arg, null_bit_arg,
+                  unireg_check_arg, field_name_arg)
 {
   /* For 4.0 MYD and 4.0 InnoDB compatibility */
-  flags|= UNSIGNED_FLAG | BINARY_FLAG;
+  flags|= UNSIGNED_FLAG;
   if (unireg_check != NONE)
   {
     /*
@@ -4636,6 +4713,7 @@ String *Field_timestamp::val_str(String *val_buffer, String *val_ptr)
 {
   MYSQL_TIME ltime;
   uint32 temp, temp2;
+  uint dec;
   char *to;
 
   val_buffer->alloc(field_length+1);
@@ -4690,6 +4768,16 @@ String *Field_timestamp::val_str(String *val_buffer, String *val_ptr)
   *to++= (char) ('0'+(char) (temp));
   *to= 0;
   val_buffer->set_charset(&my_charset_numeric);
+
+  if ((dec= decimals()))
+  {
+    ulong sec_part= (ulong) sec_part_shift(ltime.second_part, dec);
+    char *buf= const_cast<char*>(val_buffer->ptr() + MAX_DATETIME_WIDTH);
+    for (int i= dec; i > 0; i--, sec_part/= 10)
+    buf[i]= (char)(sec_part % 10) + '0';
+    buf[0]= '.';
+    buf[dec + 1]= 0;
+  }
   return val_buffer;
 }
 
@@ -4743,7 +4831,14 @@ void Field_timestamp::sort_string(uchar *to,uint length __attribute__((unused)))
 
 void Field_timestamp::sql_type(String &res) const
 {
-  res.set_ascii(STRING_WITH_LEN("timestamp"));
+  if (!decimals())
+  {
+    res.set_ascii(STRING_WITH_LEN("timestamp"));
+    return;
+  }
+  CHARSET_INFO *cs=res.charset();
+  res.length(cs->cset->snprintf(cs, (char*) res.ptr(), res.alloced_length(),
+                                "timestamp(%u)", decimals()));
 }
 
 
@@ -4783,13 +4878,6 @@ void Field_timestamp::set_explicit_default(Item *value)
   set_has_explicit_value();
 }
 
-void Field_timestamp_hires::sql_type(String &res) const
-{
-  CHARSET_INFO *cs=res.charset();
-  res.length(cs->cset->snprintf(cs, (char*) res.ptr(), res.alloced_length(),
-                                "timestamp(%u)", dec));
-}
-
 #ifdef NOT_USED
 static void store_native(ulonglong num, uchar *to, uint bytes)
 {
@@ -4883,7 +4971,7 @@ my_time_t Field_timestamp_hires::get_timestamp(ulong *sec_part) const
   return mi_uint4korr(ptr);
 }
 
-double Field_timestamp_hires::val_real(void)
+double Field_timestamp_with_dec::val_real(void)
 {
   MYSQL_TIME ltime;
   if (get_date(&ltime, TIME_NO_ZERO_DATE))
@@ -4894,31 +4982,14 @@ double Field_timestamp_hires::val_real(void)
          ltime.minute * 1e2 + ltime.second + ltime.second_part*1e-6;
 }
 
-String *Field_timestamp_hires::val_str(String *val_buffer, String *val_ptr)
-{
-  String *tmp= Field_timestamp::val_str(val_buffer, val_ptr);
-  ulong sec_part= (ulong)read_bigendian(ptr+4, sec_part_bytes[dec]);
-  
-  if (tmp->ptr() == zero_timestamp)
-    return tmp;
-
-  char *buf= const_cast<char*>(tmp->ptr() + MAX_DATETIME_WIDTH);
-  for (int i=dec; i>0; i--, sec_part/=10)
-    buf[i]= (char)(sec_part % 10) + '0';
-  buf[0]= '.';
-  buf[dec+1]= 0;
-  return tmp;
-}
-
-
-my_decimal *Field_timestamp_hires::val_decimal(my_decimal *d)
+my_decimal *Field_timestamp_with_dec::val_decimal(my_decimal *d)
 {
   MYSQL_TIME ltime;
   get_date(&ltime, 0);
   return TIME_to_my_decimal(&ltime, d);
 }
  
-int Field_timestamp_hires::store_decimal(const my_decimal *d)
+int Field_timestamp::store_decimal(const my_decimal *d)
 {
   ulonglong nr;
   ulong sec_part;
@@ -4941,7 +5012,7 @@ int Field_timestamp_hires::store_decimal(const my_decimal *d)
   return store_TIME_with_warning(thd, &ltime, &str, error, tmp != -1);
 }
 
-int Field_timestamp_hires::set_time()
+int Field_timestamp_with_dec::set_time()
 {
   THD *thd= get_thd();
   set_notnull();
@@ -4949,7 +5020,7 @@ int Field_timestamp_hires::set_time()
   return 0;
 }
 
-bool Field_timestamp_hires::send_binary(Protocol *protocol)
+bool Field_timestamp_with_dec::send_binary(Protocol *protocol)
 {
   MYSQL_TIME ltime;
   Field_timestamp::get_date(&ltime, 0);
@@ -4970,23 +5041,72 @@ int Field_timestamp_hires::cmp(const uchar *a_ptr, const uchar *b_ptr)
 }
 
 
-void Field_timestamp_hires::sort_string(uchar *to,uint length)
-{
-  DBUG_ASSERT(length == Field_timestamp_hires::pack_length());
-  memcpy(to, ptr, length);
-}
-
 uint32 Field_timestamp_hires::pack_length() const
 {
   return 4 + sec_part_bytes[dec];
 }
 
-void Field_timestamp_hires::make_field(Send_field *field)
+void Field_timestamp_with_dec::make_field(Send_field *field)
 {
   Field::make_field(field);
   field->decimals= dec;
 }
 
+
+/*************************************************************
+** MySQL-5.6 compatible TIMESTAMP(N)
+**************************************************************/
+
+void Field_timestampf::store_TIME(my_time_t timestamp, ulong sec_part)
+{
+  struct timeval tm;
+  tm.tv_sec= timestamp;
+  tm.tv_usec= sec_part;
+  my_timeval_trunc(&tm, dec);
+  my_timestamp_to_binary(&tm, ptr, dec);
+}
+
+
+my_time_t Field_timestampf::get_timestamp(ulong *sec_part) const
+{
+  struct timeval tm;
+  my_timestamp_from_binary(&tm, ptr, dec);
+  *sec_part= tm.tv_usec;
+  return tm.tv_sec;
+}
+
+
+/*************************************************************/
+uint Field_temporal::is_equal(Create_field *new_field)
+{
+  return new_field->sql_type == real_type() &&
+         new_field->length == max_display_length();
+}
+
+
+void Field_temporal::set_warnings(MYSQL_ERROR::enum_warning_level trunc_level,
+                                  const ErrConv *str, int was_cut,
+                                  timestamp_type ts_type)
+{
+  /*
+    error code logic:
+    MYSQL_TIME_WARN_TRUNCATED means that the value was not a date/time at all.
+      it will be stored as zero date/time.
+    MYSQL_TIME_WARN_OUT_OF_RANGE means that the value was a date/time,
+      that is, it was parsed as such, but the value was invalid.
+
+    Also, MYSQL_TIME_WARN_TRUNCATED is used when storing a DATETIME in
+    a DATE field and non-zero time part is thrown away.
+  */
+  if (was_cut & MYSQL_TIME_WARN_TRUNCATED)
+    set_datetime_warning(trunc_level, WARN_DATA_TRUNCATED,
+                         str, mysql_type_to_time_type(type()), 1);
+  if (was_cut & MYSQL_TIME_WARN_OUT_OF_RANGE)
+    set_datetime_warning(MYSQL_ERROR::WARN_LEVEL_WARN, ER_WARN_DATA_OUT_OF_RANGE,
+                         str, mysql_type_to_time_type(type()), 1);
+}
+
+
 /*
   Store string into a date/time field
 
@@ -4997,21 +5117,21 @@ void Field_timestamp_hires::make_field(Send_field *field)
     3  Datetime value that was cut (warning level NOTE)
        This is used by opt_range.cc:get_mm_leaf().
 */
-int Field_temporal::store_TIME_with_warning(MYSQL_TIME *ltime,
-                                            const ErrConv *str,
-                                            int was_cut, int have_smth_to_conv)
+int Field_temporal_with_date::store_TIME_with_warning(MYSQL_TIME *ltime,
+                                                      const ErrConv *str,
+                                                      int was_cut,
+                                                      int have_smth_to_conv)
 {
   MYSQL_ERROR::enum_warning_level trunc_level= MYSQL_ERROR::WARN_LEVEL_WARN;
   int ret= 2;
   
   ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
 
-  if (was_cut == 0 &&
-      have_smth_to_conv == 0 &&
-      mysql_type_to_time_type(type()) != MYSQL_TIMESTAMP_TIME) // special case: zero date
+  if (was_cut == 0 && have_smth_to_conv == 0) // special case: zero date
+  {
     was_cut= MYSQL_TIME_WARN_OUT_OF_RANGE;
-  else
-  if (!have_smth_to_conv)
+  }
+  else if (!have_smth_to_conv)
   {
     bzero(ltime, sizeof(*ltime));
     was_cut=  MYSQL_TIME_WARN_TRUNCATED;
@@ -5025,39 +5145,13 @@ int Field_temporal::store_TIME_with_warning(MYSQL_TIME *ltime,
     was_cut|=  MYSQL_TIME_WARN_TRUNCATED;
     ret= 3;
   }
-  else if (!(was_cut & MYSQL_TIME_WARN_TRUNCATED) &&
-           mysql_type_to_time_type(type()) == MYSQL_TIMESTAMP_TIME &&
-           (ltime->year || ltime->month))
-  {
-    ltime->year= ltime->month= ltime->day= 0;
-    trunc_level= MYSQL_ERROR::WARN_LEVEL_NOTE;
-    was_cut|=  MYSQL_TIME_WARN_TRUNCATED;
-    ret= 3;
-  }
-
-  /*
-    error code logic:
-    MYSQL_TIME_WARN_TRUNCATED means that the value was not a date/time at all.
-      it will be stored as zero date/time.
-    MYSQL_TIME_WARN_OUT_OF_RANGE means that the value was a date/time,
-      that is, it was parsed as such, but the value was invalid.
-
-    Also, MYSQL_TIME_WARN_TRUNCATED is used when storing a DATETIME in
-    a DATE field and non-zero time part is thrown away.
-  */
-  if (was_cut & MYSQL_TIME_WARN_TRUNCATED)
-    set_datetime_warning(trunc_level, WARN_DATA_TRUNCATED,
-                         str, mysql_type_to_time_type(type()), 1);
-  if (was_cut & MYSQL_TIME_WARN_OUT_OF_RANGE)
-    set_datetime_warning(MYSQL_ERROR::WARN_LEVEL_WARN, ER_WARN_DATA_OUT_OF_RANGE,
-                         str, mysql_type_to_time_type(type()), 1);
-
+  set_warnings(trunc_level, str, was_cut, mysql_type_to_time_type(type()));
   store_TIME(ltime);
   return was_cut ? ret : 0;
 }
 
 
-int Field_temporal::store(const char *from,uint len,CHARSET_INFO *cs)
+int Field_temporal_with_date::store(const char *from, uint len, CHARSET_INFO *cs)
 {
   MYSQL_TIME ltime;
   int error;
@@ -5075,7 +5169,7 @@ int Field_temporal::store(const char *from,uint len,CHARSET_INFO *cs)
 }
 
 
-int Field_temporal::store(double nr)
+int Field_temporal_with_date::store(double nr)
 {
   int error= 0;
   MYSQL_TIME ltime;
@@ -5092,7 +5186,7 @@ int Field_temporal::store(double nr)
 }
 
 
-int Field_temporal::store(longlong nr, bool unsigned_val)
+int Field_temporal_with_date::store(longlong nr, bool unsigned_val)
 {
   int error;
   MYSQL_TIME ltime;
@@ -5110,7 +5204,7 @@ int Field_temporal::store(longlong nr, bool unsigned_val)
 }
 
 
-int Field_temporal::store_time_dec(MYSQL_TIME *ltime, uint dec)
+int Field_temporal_with_date::store_time_dec(MYSQL_TIME *ltime, uint dec)
 {
   int error = 0, have_smth_to_conv= 1;
   MYSQL_TIME l_time= *ltime;
@@ -5144,6 +5238,35 @@ my_decimal *Field_temporal::val_decimal(my_decimal *d)
 ** In number context: HHMMSS
 ** Stored as a 3 byte unsigned int
 ****************************************************************************/
+int Field_time::store_TIME_with_warning(MYSQL_TIME *ltime,
+                                        const ErrConv *str,
+                                        int was_cut,
+                                        int have_smth_to_conv)
+{
+  MYSQL_ERROR::enum_warning_level trunc_level= MYSQL_ERROR::WARN_LEVEL_WARN;
+  int ret= 2;
+  
+  ASSERT_COLUMN_MARKED_FOR_WRITE_OR_COMPUTED;
+
+  if (!have_smth_to_conv)
+  {
+    bzero(ltime, sizeof(*ltime));
+    was_cut= MYSQL_TIME_WARN_TRUNCATED;
+    ret= 1;
+  }
+  else if (!(was_cut & MYSQL_TIME_WARN_TRUNCATED) &&
+           (ltime->year || ltime->month))
+  {
+    ltime->year= ltime->month= ltime->day= 0;
+    trunc_level= MYSQL_ERROR::WARN_LEVEL_NOTE;
+    was_cut|=  MYSQL_TIME_WARN_TRUNCATED;
+    ret= 3;
+  }
+  set_warnings(trunc_level, str, was_cut, MYSQL_TIMESTAMP_TIME);
+  store_TIME(ltime);
+  return was_cut ? ret : 0;
+}
+
 
 void Field_time::store_TIME(MYSQL_TIME *ltime)
 {
@@ -5229,32 +5352,16 @@ longlong Field_time::val_int(void)
   my_charset_bin
 */
 
-String *Field_time::val_str(String *val_buffer,
-			    String *val_ptr __attribute__((unused)))
+String *Field_time::val_str(String *str,
+			    String *unused __attribute__((unused)))
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
   MYSQL_TIME ltime;
-  long tmp=(long) sint3korr(ptr);
-  ltime.neg= 0;
-  if (tmp < 0)
-  {
-    tmp= -tmp;
-    ltime.neg= 1;
-  }
-  ltime.year= ltime.month= 0;
-  ltime.day= (uint) 0;
-  ltime.hour= (uint) (tmp/10000);
-  ltime.minute= (uint) (tmp/100 % 100);
-  ltime.second= (uint) (tmp % 100);
-  ltime.second_part= 0;
-
-  val_buffer->alloc(MAX_DATE_STRING_REP_LENGTH);
-  uint length= (uint) my_time_to_str(&ltime,
-                                     const_cast<char*>(val_buffer->ptr()), 0);
-  val_buffer->length(length);
-  val_buffer->set_charset(&my_charset_numeric);
-
-  return val_buffer;
+  get_date(&ltime, TIME_TIME_ONLY);
+  str->alloc(field_length + 1);
+  str->length(my_time_to_str(&ltime, const_cast<char*>(str->ptr()), decimals()));
+  str->set_charset(&my_charset_numeric);
+  return str;
 }
 
 
@@ -5297,8 +5404,8 @@ bool Field_time::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
 bool Field_time::send_binary(Protocol *protocol)
 {
   MYSQL_TIME ltime;
-  Field_time::get_date(&ltime, TIME_TIME_ONLY);
-  return protocol->store_time(&ltime, 0);
+  get_date(&ltime, TIME_TIME_ONLY);
+  return protocol->store_time(&ltime, decimals());
 }
 
 
@@ -5319,7 +5426,14 @@ void Field_time::sort_string(uchar *to,uint length __attribute__((unused)))
 
 void Field_time::sql_type(String &res) const
 {
-  res.set_ascii(STRING_WITH_LEN("time"));
+  if (decimals() == 0)
+  {
+    res.set_ascii(STRING_WITH_LEN("time"));
+    return;
+  }
+  const CHARSET_INFO *cs= res.charset();
+  res.length(cs->cset->snprintf(cs, (char*) res.ptr(), res.alloced_length(),
+                               "time(%d)", decimals()));
 }
 
 int Field_time_hires::reset()
@@ -5335,7 +5449,7 @@ void Field_time_hires::store_TIME(MYSQL_TIME *ltime)
   store_bigendian(packed, ptr, Field_time_hires::pack_length());
 }
 
-int Field_time_hires::store_decimal(const my_decimal *d)
+int Field_time::store_decimal(const my_decimal *d)
 {
   ulonglong nr;
   ulong sec_part;
@@ -5354,35 +5468,23 @@ uint32 Field_time_hires::pack_length() const
   return time_hires_bytes[dec];
 }
 
-longlong Field_time_hires::val_int(void)
+longlong Field_time_with_dec::val_int(void)
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
   MYSQL_TIME ltime;
-  Field_time_hires::get_date(&ltime, TIME_TIME_ONLY);
+  get_date(&ltime, TIME_TIME_ONLY);
   longlong val= TIME_to_ulonglong_time(&ltime);
   return ltime.neg ? -val : val;
 }
 
-double Field_time_hires::val_real(void)
+double Field_time_with_dec::val_real(void)
 {
   ASSERT_COLUMN_MARKED_FOR_READ;
   MYSQL_TIME ltime;
-  Field_time_hires::get_date(&ltime, TIME_TIME_ONLY);
+  get_date(&ltime, TIME_TIME_ONLY);
   return TIME_to_double(&ltime);
 }
 
-String *Field_time_hires::val_str(String *str,
-                                  String *unused __attribute__((unused)))
-{
-  ASSERT_COLUMN_MARKED_FOR_READ;
-  MYSQL_TIME ltime;
-  Field_time_hires::get_date(&ltime, TIME_TIME_ONLY);
-  str->alloc(field_length+1);
-  str->length(my_time_to_str(&ltime, (char*) str->ptr(), dec));
-  str->set_charset(&my_charset_bin);
-  return str;
-}
-
 bool Field_time_hires::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
 {
   uint32 len= pack_length();
@@ -5402,14 +5504,6 @@ bool Field_time_hires::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
 }
 
 
-bool Field_time_hires::send_binary(Protocol *protocol)
-{
-  MYSQL_TIME ltime;
-  Field_time_hires::get_date(&ltime, TIME_TIME_ONLY);
-  return protocol->store_time(&ltime, dec);
-}
-
-
 int Field_time_hires::cmp(const uchar *a_ptr, const uchar *b_ptr)
 {
   ulonglong a=read_bigendian(a_ptr, Field_time_hires::pack_length());
@@ -5424,17 +5518,36 @@ void Field_time_hires::sort_string(uchar *to,uint length __attribute__((unused))
   to[0]^= 128;
 }
 
-void Field_time_hires::sql_type(String &res) const
+void Field_time_with_dec::make_field(Send_field *field)
 {
-  CHARSET_INFO *cs=res.charset();
-  res.length(cs->cset->snprintf(cs, (char*) res.ptr(), res.alloced_length(),
-                                "time(%u)", dec));
+  Field::make_field(field);
+  field->decimals= dec;
 }
 
-void Field_time_hires::make_field(Send_field *field)
+/****************************************************************************
+** time type with fsp (MySQL-5.6 version)
+** In string context: HH:MM:SS.FFFFFF
+** In number context: HHMMSS.FFFFFF
+****************************************************************************/
+
+int Field_timef::reset()
 {
-  Field::make_field(field);
-  field->decimals= dec;
+  my_time_packed_to_binary(0, ptr, dec);
+  return 0;
+}
+
+void Field_timef::store_TIME(MYSQL_TIME *ltime)
+{
+  my_time_trunc(ltime, decimals());
+  longlong tmp= TIME_to_longlong_time_packed(ltime);
+  my_time_packed_to_binary(tmp, ptr, dec);
+}
+
+bool Field_timef::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
+{
+  longlong tmp= my_time_packed_from_binary(ptr, dec);
+  TIME_from_longlong_time_packed(ltime, tmp);
+  return false;
 }
 
 /****************************************************************************
@@ -5792,8 +5905,8 @@ bool Field_datetime::send_binary(Protocol *protocol)
   Field_datetime::get_date(&tm, TIME_FUZZY_DATE);
   return protocol->store(&tm, 0);
 }
-  
-  
+
+
 double Field_datetime::val_real(void)
 {
   return (double) Field_datetime::val_int();
@@ -5901,7 +6014,14 @@ void Field_datetime::sort_string(uchar *to,uint length __attribute__((unused)))
 
 void Field_datetime::sql_type(String &res) const
 {
-  res.set_ascii(STRING_WITH_LEN("datetime"));
+  if (decimals() == 0)
+  {
+    res.set_ascii(STRING_WITH_LEN("datetime"));
+    return;
+  }
+  CHARSET_INFO *cs= res.charset();
+  res.length(cs->cset->snprintf(cs, (char*) res.ptr(), res.alloced_length(),
+                                "datetime(%u)", decimals()));
 }
 
 
@@ -5924,7 +6044,7 @@ void Field_datetime_hires::store_TIME(MYSQL_TIME *ltime)
   store_bigendian(packed, ptr, Field_datetime_hires::pack_length());
 }
 
-int Field_datetime_hires::store_decimal(const my_decimal *d)
+int Field_temporal_with_date::store_decimal(const my_decimal *d)
 {
   ulonglong nr;
   ulong sec_part;
@@ -5949,38 +6069,38 @@ int Field_datetime_hires::store_decimal(const my_decimal *d)
   return store_TIME_with_warning(&ltime, &str, error, tmp != -1);
 }
 
-bool Field_datetime_hires::send_binary(Protocol *protocol)
+bool Field_datetime_with_dec::send_binary(Protocol *protocol)
 {
   MYSQL_TIME ltime;
-  Field_datetime_hires::get_date(&ltime, TIME_FUZZY_DATE);
+  get_date(&ltime, TIME_FUZZY_DATE);
   return protocol->store(&ltime, dec);
 }
 
 
-double Field_datetime_hires::val_real(void)
+double Field_datetime_with_dec::val_real(void)
 {
   MYSQL_TIME ltime;
-  Field_datetime_hires::get_date(&ltime, TIME_FUZZY_DATE);
+  get_date(&ltime, TIME_FUZZY_DATE);
   return TIME_to_double(&ltime);
 }
 
-longlong Field_datetime_hires::val_int(void)
+longlong Field_datetime_with_dec::val_int(void)
 {
   MYSQL_TIME ltime;
-  Field_datetime_hires::get_date(&ltime, TIME_FUZZY_DATE);
+  get_date(&ltime, TIME_FUZZY_DATE);
   return TIME_to_ulonglong_datetime(&ltime);
 }
 
 
-String *Field_datetime_hires::val_str(String *str,
-                                      String *unused __attribute__((unused)))
+String *Field_datetime_with_dec::val_str(String *str,
+                                         String *unused __attribute__((unused)))
 {
   MYSQL_TIME ltime;
-  Field_datetime_hires::get_date(&ltime, TIME_FUZZY_DATE);
+  get_date(&ltime, TIME_FUZZY_DATE);
   str->alloc(field_length+1);
   str->length(field_length);
   my_datetime_to_str(&ltime, (char*) str->ptr(), dec);
-  str->set_charset(&my_charset_bin);
+  str->set_charset(&my_charset_numeric);
   return str;
 }
 
@@ -6007,27 +6127,42 @@ int Field_datetime_hires::cmp(const uchar *a_ptr, const uchar *b_ptr)
   return a < b ? -1 : a > b ? 1 : 0;
 }
 
-void Field_datetime_hires::sort_string(uchar *to,
-                                       uint length __attribute__((unused)))
+void Field_datetime_with_dec::make_field(Send_field *field)
 {
-  DBUG_ASSERT(length == Field_datetime_hires::pack_length());
-  memcpy(to, ptr, length);
+  Field::make_field(field);
+  field->decimals= dec;
 }
 
 
-void Field_datetime_hires::sql_type(String &res) const
+/****************************************************************************
+** MySQL-5.6 compatible DATETIME(N)
+**
+****************************************************************************/
+int Field_datetimef::reset()
 {
-  CHARSET_INFO *cs=res.charset();
-  res.length(cs->cset->snprintf(cs, (char*) res.ptr(), res.alloced_length(),
-                                "datetime(%u)", dec));
+  my_datetime_packed_to_binary(0, ptr, dec);
+  return 0;
 }
 
-void Field_datetime_hires::make_field(Send_field *field)
+void Field_datetimef::store_TIME(MYSQL_TIME *ltime)
 {
-  Field::make_field(field);
-  field->decimals= dec;
+  my_time_trunc(ltime, decimals());
+  longlong tmp= TIME_to_longlong_datetime_packed(ltime);
+  my_datetime_packed_to_binary(tmp, ptr, dec);
+}
+
+bool Field_datetimef::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
+{
+  longlong tmp= my_datetime_packed_from_binary(ptr, dec);
+  TIME_from_longlong_datetime_packed(ltime, tmp);
+  if (!tmp)
+    return fuzzydate & TIME_NO_ZERO_DATE;
+  if (!ltime->month || !ltime->day)
+    return !(fuzzydate & TIME_FUZZY_DATE);
+  return false;
 }
 
+
 /****************************************************************************
 ** string type
 ** A string may be varchar or binary
@@ -6198,80 +6333,6 @@ int Field_str::store(double nr)
   return store(buff, length, &my_charset_numeric);
 }
 
-static
-inline ulonglong char_prefix_to_ulonglong(uchar *src)
-{
-  uint sz= sizeof(ulonglong);
-  for (uint i= 0; i < sz/2; i++)
-  {
-    uchar tmp= src[i];
-    src[i]= src[sz-1-i];
-    src[sz-1-i]= tmp;
-  }
-  return uint8korr(src); 
-}
-
-/**
-  @brief
-  Determine the relative position of the field value in a string interval
-
-  @details
-  The function returns a double number between 0.0 and 1.0 as the relative
-  position of the value of the this field in the string interval of [min,max].
-  If the value is not in the interval the the function returns 0.0 when
-  the value is less than min, and, 1.0 when the value is greater than max.
-
-  @note
-  To calculate the relative position of the string value v in the interval
-  [min, max] the function first converts the beginning of these three
-  strings v, min, max into the strings that are used for byte comparison.
-  For each string not more sizeof(ulonglong) first bytes are taken
-  from the result of conversion. Then these bytes are interpreted as the
-  big-endian representation of an ulonglong integer. The values of these
-  integer numbers obtained for the strings v, min, max are used to calculate
-  the position of v in [min,max] in the same way is it's done for numeric
-  fields (see Field_num::pos_in_interval).
-
-  @todo
-  Improve the procedure for the case when min and max have the same
-  beginning
-     
-  @param  min  value of the left end of the interval
-  @param  max  value of the right end of the interval
-
-  @return
-  relative position of the field value in the string interval [min,max] 
-*/
-
-double Field_str::pos_in_interval(Field *min, Field *max)
-{
-  uchar mp_prefix[sizeof(ulonglong)];
-  uchar minp_prefix[sizeof(ulonglong)];
-  uchar maxp_prefix[sizeof(ulonglong)];
-  ulonglong mp, minp, maxp;
-  my_strnxfrm(charset(), mp_prefix, sizeof(mp),
-              ptr + length_size(),
-              data_length());
-  my_strnxfrm(charset(), minp_prefix, sizeof(minp),
-              min->ptr + length_size(),
-              min->data_length());
-  my_strnxfrm(charset(), maxp_prefix, sizeof(maxp),
-              max->ptr + length_size(),
-              max->data_length());
-  mp= char_prefix_to_ulonglong(mp_prefix);
-  minp= char_prefix_to_ulonglong(minp_prefix);
-  maxp= char_prefix_to_ulonglong(maxp_prefix);
-  double n, d;
-  n= mp - minp;
-  if (n < 0)
-    return 0.0;
-  d= maxp - minp;
-  if (d <= 0)
-    return 1.0;
-  return min(n/d, 1.0);
-}
-
-
 uint Field::is_equal(Create_field *new_field)
 {
   return (new_field->sql_type == real_type());
@@ -8482,36 +8543,6 @@ my_decimal *Field_bit::val_decimal(my_decimal *deciaml_value)
 }
 
 
-/**
-  @brief
-  Determine the relative position of the field value in a bit interval
-
-  @details
-  The function returns a double number between 0.0 and 1.0 as the relative
-  position of the value of the this field in the bit interval of [min,max].
-  If the value is not in the interval the the function returns 0.0 when
-  the value is less than min, and, 1.0 when the value is greater than max.
-
-  @param  min  value of the left end of the interval
-  @param  max  value of the right end of the interval
-
-  @return
-  relative position of the field value in the bit interval [min,max] 
-*/
-
-double Field_bit::pos_in_interval(Field *min, Field *max)
-{
-  double n, d;
-  n= val_real() - min->val_real();
-  if (n < 0)
-    return 0.0;
-  d= max->val_real() - min->val_real();
-  if (d <= 0)
-    return 1.0;
-  return min(n/d, 1.0);
-}
-
-
 /*
   Compare two bit fields using pointers within the record.
   SYNOPSIS
@@ -9144,7 +9175,7 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
     it is NOT NULL, not an AUTO_INCREMENT field and not a TIMESTAMP.
   */
   if (!fld_default_value && !(fld_type_modifier & AUTO_INCREMENT_FLAG) &&
-      (fld_type_modifier & NOT_NULL_FLAG) && fld_type != MYSQL_TYPE_TIMESTAMP)
+      (fld_type_modifier & NOT_NULL_FLAG) && !is_timestamp_type(fld_type))
     flags|= NO_DEFAULT_VALUE_FLAG;
 
   if (fld_length != NULL)
@@ -9306,6 +9337,7 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
     }
     break;
   case MYSQL_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_TIMESTAMP2:
     if (length > MAX_DATETIME_PRECISION)
     {
       my_error(ER_TOO_BIG_PRECISION, MYF(0), length, fld_name,
@@ -9323,6 +9355,7 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
     length= MAX_DATE_WIDTH;
     break;
   case MYSQL_TYPE_TIME:
+  case MYSQL_TYPE_TIME2:
     if (length > MAX_DATETIME_PRECISION)
     {
       my_error(ER_TOO_BIG_PRECISION, MYF(0), length, fld_name,
@@ -9332,6 +9365,7 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
     length+= MIN_TIME_WIDTH + (length ? 1 : 0);
     break;
   case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_DATETIME2:
     if (length > MAX_DATETIME_PRECISION)
     {
       my_error(ER_TOO_BIG_PRECISION, MYF(0), length, fld_name,
@@ -9413,17 +9447,6 @@ bool Create_field::init(THD *thd, char *fld_name, enum_field_types fld_type,
     DBUG_RETURN(TRUE);
   }
 
-  switch (fld_type) {
-  case MYSQL_TYPE_DATE:
-  case MYSQL_TYPE_NEWDATE:
-  case MYSQL_TYPE_TIME:
-  case MYSQL_TYPE_DATETIME:
-  case MYSQL_TYPE_TIMESTAMP:
-    charset= &my_charset_numeric;
-    flags|= BINARY_FLAG;
-  default: break;
-  }
-
   DBUG_RETURN(FALSE); /* success */
 }
 
@@ -9462,10 +9485,16 @@ uint32 calc_pack_length(enum_field_types type,uint32 length)
   case MYSQL_TYPE_TIME:   return length > MIN_TIME_WIDTH
                             ? time_hires_bytes[length - 1 - MIN_TIME_WIDTH]
                             : 3;
+  case MYSQL_TYPE_TIME2:
+    return length > MIN_TIME_WIDTH ?
+           my_time_binary_length(length - MIN_TIME_WIDTH - 1) : 3;
   case MYSQL_TYPE_TIMESTAMP:
                           return length > MAX_DATETIME_WIDTH
                             ? 4 + sec_part_bytes[length - 1 - MAX_DATETIME_WIDTH]
                             : 4;
+  case MYSQL_TYPE_TIMESTAMP2:
+    return length > MAX_DATETIME_WIDTH ?
+           my_timestamp_binary_length(length - MAX_DATETIME_WIDTH - 1) : 4;
   case MYSQL_TYPE_DATE:
   case MYSQL_TYPE_LONG	: return 4;
   case MYSQL_TYPE_FLOAT : return sizeof(float);
@@ -9474,6 +9503,9 @@ uint32 calc_pack_length(enum_field_types type,uint32 length)
                           return length > MAX_DATETIME_WIDTH
                             ? datetime_hires_bytes[length - 1 - MAX_DATETIME_WIDTH]
                             : 8;
+  case MYSQL_TYPE_DATETIME2:
+    return length > MAX_DATETIME_WIDTH ?
+           my_datetime_binary_length(length - MAX_DATETIME_WIDTH - 1) : 5;
   case MYSQL_TYPE_LONGLONG: return 8;	/* Don't crash if no longlong */
   case MYSQL_TYPE_NULL	: return 0;
   case MYSQL_TYPE_TINY_BLOB:	return 1+portable_sizeof_char_ptr;
@@ -9538,16 +9570,6 @@ Field *make_field(TABLE_SHARE *share, uchar *ptr, uint32 field_length,
     null_bit= ((uchar) 1) << null_bit;
   }
 
-  switch (field_type) {
-  case MYSQL_TYPE_DATE:
-  case MYSQL_TYPE_NEWDATE:
-  case MYSQL_TYPE_TIME:
-  case MYSQL_TYPE_DATETIME:
-  case MYSQL_TYPE_TIMESTAMP:
-    field_charset= &my_charset_numeric;
-  default: break;
-  }
-
   DBUG_PRINT("debug", ("field_type: %d, field_length: %u, interval: %p, pack_flag: %s%s%s%s%s",
                        field_type, field_length, interval,
                        FLAGSTR(pack_flag, FIELDFLAG_BINARY),
@@ -9661,30 +9683,51 @@ Field *make_field(TABLE_SHARE *share, uchar *ptr, uint32 field_length,
     uint dec= field_length > MAX_DATETIME_WIDTH ?
                        field_length - MAX_DATETIME_WIDTH - 1: 0;
     return new_Field_timestamp(ptr, null_pos, null_bit, unireg_check,
-                               field_name, share, dec, field_charset);
+                               field_name, share, dec);
+  }
+  case MYSQL_TYPE_TIMESTAMP2:
+  {
+    uint dec= field_length > MAX_DATETIME_WIDTH ?
+                       field_length - MAX_DATETIME_WIDTH - 1: 0;
+    return new Field_timestampf(ptr, null_pos, null_bit, unireg_check,
+                                field_name, share, dec);
   }
   case MYSQL_TYPE_YEAR:
     return new Field_year(ptr,field_length,null_pos,null_bit,
 			  unireg_check, field_name);
   case MYSQL_TYPE_DATE:
     return new Field_date(ptr,null_pos,null_bit,
-			  unireg_check, field_name, field_charset);
+                          unireg_check, field_name);
   case MYSQL_TYPE_NEWDATE:
     return new Field_newdate(ptr,null_pos,null_bit,
-			     unireg_check, field_name, field_charset);
+                             unireg_check, field_name);
   case MYSQL_TYPE_TIME:
   {
     uint dec= field_length > MIN_TIME_WIDTH ?
                        field_length - MIN_TIME_WIDTH - 1: 0;
     return new_Field_time(ptr, null_pos, null_bit, unireg_check,
-                              field_name, dec, field_charset);
+                          field_name, dec);
+  }
+  case MYSQL_TYPE_TIME2:
+  {
+    uint dec= field_length > MIN_TIME_WIDTH ?
+                       field_length - MIN_TIME_WIDTH - 1: 0;
+    return new Field_timef(ptr, null_pos, null_bit, unireg_check,
+                           field_name, dec);
   }
   case MYSQL_TYPE_DATETIME:
   {
     uint dec= field_length > MAX_DATETIME_WIDTH ?
                        field_length - MAX_DATETIME_WIDTH - 1: 0;
     return new_Field_datetime(ptr, null_pos, null_bit, unireg_check,
-                              field_name, dec, field_charset);
+                              field_name, dec);
+  }
+  case MYSQL_TYPE_DATETIME2:
+  {
+    uint dec= field_length > MAX_DATETIME_WIDTH ?
+                       field_length - MAX_DATETIME_WIDTH - 1: 0;
+    return new Field_datetimef(ptr, null_pos, null_bit, unireg_check,
+                              field_name, dec);
   }
   case MYSQL_TYPE_NULL:
     return new Field_null(ptr, field_length, unireg_check, field_name,
diff --git a/sql/field.h b/sql/field.h
index 48d873beb32..feef0cbef08 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -30,6 +30,7 @@
 #include "sql_string.h"                         /* String */
 #include "my_decimal.h"                         /* my_decimal */
 #include "sql_error.h"                          /* MYSQL_ERROR */
+#include "compat56.h"
 
 class Send_field;
 class Protocol;
@@ -89,6 +90,42 @@ inline uint get_set_pack_length(int elements)
   return len > 4 ? 8 : len;
 }
 
+
+/**
+   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().
+  
+  @param real_type  Real type.
+  @retval           Field type.
+*/
+inline enum_field_types real_type_to_type(enum_field_types real_type)
+{
+  switch (real_type)
+  {
+  case MYSQL_TYPE_TIME2:
+    return MYSQL_TYPE_TIME;
+  case MYSQL_TYPE_DATETIME2:
+    return MYSQL_TYPE_DATETIME;
+  case MYSQL_TYPE_TIMESTAMP2:
+    return MYSQL_TYPE_TIMESTAMP;
+  case MYSQL_TYPE_NEWDATE:
+    return MYSQL_TYPE_DATE;
+  /* Note: NEWDECIMAL is a type, not only a real_type */
+  default: return real_type;
+  }
+}
+
+
 /*
   Virtual_column_info is the class to contain additional
   characteristics that is specific for a virtual/computed
@@ -428,6 +465,54 @@ public:
   virtual uint32 key_length() const { return pack_length(); }
   virtual enum_field_types type() const =0;
   virtual enum_field_types real_type() const { return type(); }
+  virtual enum_field_types binlog_type() const
+  {
+    /*
+      Binlog stores field->type() as type code by default. For example,
+      it puts MYSQL_TYPE_STRING in case of CHAR, VARCHAR, SET and ENUM,
+      with extra data type details put into metadata.
+
+      Binlog behaviour slightly differs between various MySQL and MariaDB
+      versions for the temporal data types TIME, DATETIME and TIMESTAMP.
+
+      MySQL prior to 5.6 uses MYSQL_TYPE_TIME, MYSQL_TYPE_DATETIME 
+      and MYSQL_TYPE_TIMESTAMP type codes in binlog and stores no 
+      additional metadata.
+
+      MariaDB-5.3 implements new versions for TIME, DATATIME, TIMESTAMP
+      with fractional second precision, but uses the old format for the
+      types TIME(0), DATETIME(0), TIMESTAMP(0), and it still stores
+      MYSQL_TYPE_TIME, MYSQL_TYPE_DATETIME and MYSQL_TYPE_TIMESTAMP in binlog,
+      with no additional metadata.
+      So row-based replication between temporal data types of
+      different precision is not possible in MariaDB.
+
+      MySQL-5.6 also implements a new version of TIME, DATETIME, TIMESTAMP
+      which support fractional second precision 0..6, and use the new
+      format even for the types TIME(0), DATETIME(0), TIMESTAMP(0).
+      For these new data types, MySQL-5.6 stores new type codes 
+      MYSQL_TYPE_TIME2, MYSQL_TYPE_DATETIME2, MYSQL_TYPE_TIMESTAMP2 in binlog,
+      with fractional precision 0..6 put into metadata.
+      This makes it in theory possible to do row-based replication between
+      columns of different fractional precision (e.g. from TIME(1) on master
+      to TIME(6) on slave). However, it's not currently fully implemented yet.
+      MySQL-5.6 can only do row-based replication from the old types
+      TIME, DATETIME, TIMESTAMP (represented by MYSQL_TYPE_TIME,
+      MYSQL_TYPE_DATETIME and MYSQL_TYPE_TIMESTAMP type codes in binlog)
+      to the new corresponding types TIME(0), DATETIME(0), TIMESTAMP(0).
+
+      Note: MariaDB starting from the version 10.0 understands the new
+      MySQL-5.6 type codes MYSQL_TYPE_TIME2, MYSQL_TYPE_DATETIME2,
+      MYSQL_TYPE_TIMESTAMP2. When started over MySQL-5.6 tables both on
+      master and on slave, MariaDB-10.0 can also do row-based replication
+      from the old types TIME, DATETIME, TIMESTAMP to the new MySQL-5.6
+      types TIME(0), DATETIME(0), TIMESTAMP(0).
+
+      Note: perhaps binlog should eventually be modified to store
+      real_type() instead of type() for all column types.
+    */
+    return type();
+  }
   inline  int cmp(const uchar *str) { return cmp(ptr,str); }
   virtual int cmp_max(const uchar *a, const uchar *b, uint max_len)
     { return cmp(a, b); }
@@ -661,6 +746,16 @@ public:
   { return binary() ? &my_charset_bin : charset(); }
   virtual CHARSET_INFO *sort_charset(void) const { return charset(); }
   virtual bool has_charset(void) const { return FALSE; }
+  /*
+    match_collation_to_optimize_range() is to distinguish in
+    range optimizer (see opt_range.cc) between real string types:
+      CHAR, VARCHAR, TEXT
+    and the other string-alike types with result_type() == STRING_RESULT:
+      DATE, TIME, DATETIME, TIMESTAMP
+    We need it to decide whether to test if collation of the operation
+    matches collation of the field (needed only for real string types).
+  */
+  virtual bool match_collation_to_optimize_range() const { return false; }
   virtual void set_charset(CHARSET_INFO *charset_arg) { }
   virtual enum Derivation derivation(void) const
   { return DERIVATION_IMPLICIT; }
@@ -808,7 +903,8 @@ protected:
   {
     return (flags & (BINCMP_FLAG | BINARY_FLAG)) != 0;
   }
-
+  double pos_in_interval_val_real(Field *min, Field *max);
+  double pos_in_interval_val_str(Field *min, Field *max, uint data_offset);
 };
 
 
@@ -847,7 +943,10 @@ public:
   bool get_int(CHARSET_INFO *cs, const char *from, uint len, 
                longlong *rnd, ulonglong unsigned_max, 
                longlong signed_min, longlong signed_max);
-  double pos_in_interval(Field *min, Field *max);
+  double pos_in_interval(Field *min, Field *max)
+  {
+    return pos_in_interval_val_real(min, max);
+  }
 };
 
 
@@ -860,23 +959,11 @@ public:
 	    uchar null_bit_arg, utype unireg_check_arg,
 	    const char *field_name_arg, CHARSET_INFO *charset);
   Item_result result_type () const { return STRING_RESULT; }
-  /*
-    match_collation_to_optimize_range() is to distinguish in
-    range optimizer (see opt_range.cc) between real string types:
-      CHAR, VARCHAR, TEXT
-    and the other string-alike types with result_type() == STRING_RESULT:
-      DATE, TIME, DATETIME, TIMESTAMP
-    We need it to decide whether to test if collation of the operation
-    matches collation of the field (needed only for real string types).
-    QQ: shouldn't DATE/TIME types have their own XXX_RESULT types eventually?
-  */
-  virtual bool match_collation_to_optimize_range() const=0;
   uint decimals() const { return NOT_FIXED_DEC; }
   int  store(double nr);
   int  store(longlong nr, bool unsigned_val)=0;
   int  store_decimal(const my_decimal *);
   int  store(const char *to,uint length,CHARSET_INFO *cs)=0;
-  uint size_of() const { return sizeof(*this); }
   uint repertoire(void) const
   {
     return my_charset_repertoire(field_charset);
@@ -894,7 +981,10 @@ public:
   uint is_equal(Create_field *new_field);
   bool eq_cmp_as_binary() { return test(flags & BINARY_FLAG); }
   virtual uint length_size() { return 0; }
-  double pos_in_interval(Field *min, Field *max);
+  double pos_in_interval(Field *min, Field *max)
+  {
+    return pos_in_interval_val_str(min, max, length_size());
+  }
 };
 
 /* base class for Field_string, Field_varstring and Field_blob */
@@ -914,6 +1004,7 @@ public:
 
   int store_decimal(const my_decimal *d);
   uint32 max_data_length() const;
+  bool match_collation_to_optimize_range() const { return true; }
 };
 
 /* base class for float and double and decimal (old one) */
@@ -1323,7 +1414,6 @@ public:
 	       unireg_check_arg, field_name_arg, cs)
     {}
   enum_field_types type() const { return MYSQL_TYPE_NULL;}
-  bool match_collation_to_optimize_range() const { return FALSE; }
   int  store(const char *to, uint length, CHARSET_INFO *cs)
   { null[0]=1; return 0; }
   int store(double nr)   { null[0]=1; return 0; }
@@ -1345,7 +1435,67 @@ public:
 };
 
 
-class Field_timestamp :public Field_str {
+class Field_temporal: public Field {
+public:
+  Field_temporal(uchar *ptr_arg,uint32 len_arg, uchar *null_ptr_arg,
+                 uchar null_bit_arg, utype unireg_check_arg,
+                 const char *field_name_arg)
+    :Field(ptr_arg, len_arg, null_ptr_arg, null_bit_arg, unireg_check_arg,
+               field_name_arg)
+    { flags|= BINARY_FLAG; }
+  Item_result result_type () const { return STRING_RESULT; }   
+  uint32 max_display_length() { return field_length; }
+  bool str_needs_quotes() { return TRUE; }
+  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
+  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
+  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
+  const CHARSET_INFO *sort_charset(void) const { return &my_charset_bin; }
+  bool binary() const { return true; }
+  enum Item_result cmp_type () const { return TIME_RESULT; }
+  uint is_equal(Create_field *new_field);
+  bool eq_def(Field *field)
+  {
+    return (Field::eq_def(field) && decimals() == field->decimals());
+  }
+  my_decimal *val_decimal(my_decimal*);
+  void set_warnings(MYSQL_ERROR::enum_warning_level trunc_level,
+                    const ErrConv *str, int was_cut, timestamp_type ts_type);
+  double pos_in_interval(Field *min, Field *max)
+  {
+    return pos_in_interval_val_real(min, max);
+  }
+};
+
+
+/**
+  Abstract class for:
+  - DATE
+  - DATETIME
+  - DATETIME(1..6)
+  - DATETIME(0..6) - MySQL56 version
+*/
+class Field_temporal_with_date: public Field_temporal {
+protected:
+  int store_TIME_with_warning(MYSQL_TIME *ltime, const ErrConv *str,
+                              int was_cut, int have_smth_to_conv);
+  virtual void store_TIME(MYSQL_TIME *ltime) = 0;
+public:
+  Field_temporal_with_date(uchar *ptr_arg, uint32 len_arg,
+                           uchar *null_ptr_arg, uchar null_bit_arg,
+                           utype unireg_check_arg,
+                           const char *field_name_arg)
+    :Field_temporal(ptr_arg, len_arg, null_ptr_arg, null_bit_arg,
+                    unireg_check_arg, field_name_arg)
+    {}
+  int  store(const char *to, uint length, CHARSET_INFO *charset);
+  int  store(double nr);
+  int  store(longlong nr, bool unsigned_val);
+  int  store_time_dec(MYSQL_TIME *ltime, uint dec);
+  int  store_decimal(const my_decimal *);
+};
+
+
+class Field_timestamp :public Field_temporal {
 protected:
   int store_TIME_with_warning(THD *, MYSQL_TIME *, const ErrConv *,
                               bool, bool);
@@ -1353,21 +1503,14 @@ public:
   Field_timestamp(uchar *ptr_arg, uint32 len_arg,
                   uchar *null_ptr_arg, uchar null_bit_arg,
 		  enum utype unireg_check_arg, const char *field_name_arg,
-		  TABLE_SHARE *share, CHARSET_INFO *cs);
-  Field_timestamp(bool maybe_null_arg, const char *field_name_arg,
-		  CHARSET_INFO *cs);
+		  TABLE_SHARE *share);
   enum_field_types type() const { return MYSQL_TYPE_TIMESTAMP;}
-  bool match_collation_to_optimize_range() const { return FALSE; }
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONG_INT; }
-  enum Item_result cmp_type () const { return TIME_RESULT; }
-  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
-  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
-  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
-  bool binary() const { return 1; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
   int  store(double nr);
   int  store(longlong nr, bool unsigned_val);
   int  store_time_dec(MYSQL_TIME *ltime, uint dec);
+  int  store_decimal(const my_decimal *);
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -1377,7 +1520,6 @@ public:
   uint32 pack_length() const { return 4; }
   void sql_type(String &str) const;
   bool zero_pack() const { return 0; }
-  uint decimals() const { return 0; }
   virtual int set_time();
   virtual void set_default()
   {
@@ -1418,46 +1560,109 @@ public:
   {
     return unpack_int32(to, from, from_end);
   }
+  uint size_of() const { return sizeof(*this); }
 };
 
 
-class Field_timestamp_hires :public Field_timestamp {
+/**
+  Abstract class for:
+  - TIMESTAMP(1..6)
+  - TIMESTAMP(0..6) - MySQL56 version
+*/
+class Field_timestamp_with_dec :public Field_timestamp {
+protected:
   uint dec;
 public:
-  Field_timestamp_hires(uchar *ptr_arg,
-                  uchar *null_ptr_arg, uchar null_bit_arg,
-                  enum utype unireg_check_arg, const char *field_name_arg,
-                  TABLE_SHARE *share, uint dec_arg, CHARSET_INFO *cs) :
-  Field_timestamp(ptr_arg, MAX_DATETIME_WIDTH + dec_arg + 1, null_ptr_arg,
-                  null_bit_arg, unireg_check_arg, field_name_arg, share, cs),
+  Field_timestamp_with_dec(uchar *ptr_arg,
+                           uchar *null_ptr_arg, uchar null_bit_arg,
+                           enum utype unireg_check_arg,
+                           const char *field_name_arg,
+                           TABLE_SHARE *share, uint dec_arg) :
+  Field_timestamp(ptr_arg,
+                  MAX_DATETIME_WIDTH + dec_arg + test(dec_arg), null_ptr_arg,
+                  null_bit_arg, unireg_check_arg, field_name_arg, share),
   dec(dec_arg)
   {
-    DBUG_ASSERT(dec);
     DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
   }
-  void sql_type(String &str) const;
-  my_time_t get_timestamp(ulong *sec_part) const;
-  void store_TIME(my_time_t timestamp, ulong sec_part);
-  int store_decimal(const my_decimal *d);
-  double val_real(void);
-  String *val_str(String*,String *);
-  my_decimal* val_decimal(my_decimal*);
-  bool send_binary(Protocol *protocol);
-  int cmp(const uchar *,const uchar *);
-  void sort_string(uchar *buff,uint length);
   uint decimals() const { return dec; }
-  int set_time();
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_BINARY; }
-  void make_field(Send_field *field);
-  uint32 pack_length() const;
   uchar *pack(uchar *to, const uchar *from, uint max_length)
   { return Field::pack(to, from, max_length); }
   const uchar *unpack(uchar* to, const uchar *from, const uchar *from_end,
                       uint param_data)
   { return Field::unpack(to, from, from_end, param_data); }
+  void make_field(Send_field *field);
+  void sort_string(uchar *to, uint length)
+  {
+    DBUG_ASSERT(length == pack_length());
+    memcpy(to, ptr, length);
+  }
+  bool send_binary(Protocol *protocol);
+  double val_real(void);
+  my_decimal* val_decimal(my_decimal*);
+  int set_time();
+};
+
+
+class Field_timestamp_hires :public Field_timestamp_with_dec {
+public:
+  Field_timestamp_hires(uchar *ptr_arg,
+                        uchar *null_ptr_arg, uchar null_bit_arg,
+                        enum utype unireg_check_arg,
+                        const char *field_name_arg,
+                        TABLE_SHARE *share, uint dec_arg) :
+  Field_timestamp_with_dec(ptr_arg, null_ptr_arg, null_bit_arg,
+                           unireg_check_arg, field_name_arg, share, dec_arg)
+  {
+    DBUG_ASSERT(dec);
+  }
+  my_time_t get_timestamp(ulong *sec_part) const;
+  void store_TIME(my_time_t timestamp, ulong sec_part);
+  int cmp(const uchar *,const uchar *);
+  uint32 pack_length() const;
+  uint size_of() const { return sizeof(*this); }
+};
+
+
+/**
+  TIMESTAMP(0..6) - MySQL56 version
+*/
+class Field_timestampf :public Field_timestamp_with_dec {
+  int do_save_field_metadata(uchar *metadata_ptr)
+  {
+    *metadata_ptr= decimals();
+    return 1;
+  }
+public:
+  Field_timestampf(uchar *ptr_arg,
+                   uchar *null_ptr_arg, uchar null_bit_arg,
+                   enum utype unireg_check_arg,
+                   const char *field_name_arg,
+                   TABLE_SHARE *share, uint dec_arg) :
+    Field_timestamp_with_dec(ptr_arg, null_ptr_arg, null_bit_arg,
+                             unireg_check_arg, field_name_arg, share, dec_arg)
+    {}
+  enum_field_types real_type() const { return MYSQL_TYPE_TIMESTAMP2; }
+  enum_field_types binlog_type() const { return MYSQL_TYPE_TIMESTAMP2; }
+  uint32 pack_length() const
+  {
+    return my_timestamp_binary_length(dec);
+  }
+  uint row_pack_length() { return pack_length(); }
+  uint pack_length_from_metadata(uint field_metadata)
+  {
+    DBUG_ENTER("Field_timestampf::pack_length_from_metadata");
+    uint tmp= my_timestamp_binary_length(field_metadata);
+    DBUG_RETURN(tmp);
+  }
+  int cmp(const uchar *a_ptr,const uchar *b_ptr)
+  {
+    return memcmp(a_ptr, b_ptr, pack_length());
+  }
+  void store_TIME(my_time_t timestamp, ulong sec_part);
+  my_time_t get_timestamp(ulong *sec_part) const;
   uint size_of() const { return sizeof(*this); }
-  bool eq_def(Field *field)
-  { return Field_str::eq_def(field) && dec == field->decimals(); }
 };
 
 
@@ -1484,56 +1689,24 @@ public:
 };
 
 
-class Field_temporal: public Field_str {
-protected:
-  int store_TIME_with_warning(MYSQL_TIME *ltime, const ErrConv *str,
-                              int was_cut, int have_smth_to_conv);
-  virtual void store_TIME(MYSQL_TIME *ltime) = 0;
-public:
-  Field_temporal(uchar *ptr_arg,uint32 len_arg, uchar *null_ptr_arg,
-                 uchar null_bit_arg, utype unireg_check_arg,
-                 const char *field_name_arg, CHARSET_INFO *charset_arg)
-    :Field_str(ptr_arg, len_arg, null_ptr_arg, null_bit_arg, unireg_check_arg,
-               field_name_arg, charset_arg)
-    { flags|= BINARY_FLAG; }
-  enum Derivation derivation(void) const { return DERIVATION_NUMERIC; }
-  uint repertoire(void) const { return MY_REPERTOIRE_NUMERIC; }
-  CHARSET_INFO *charset(void) const { return &my_charset_numeric; }
-  bool binary() const { return 1; }
-  bool match_collation_to_optimize_range() const { return FALSE; }
-  enum Item_result cmp_type () const { return TIME_RESULT; }
-  int  store(const char *to,uint length,CHARSET_INFO *charset);
-  int  store(double nr);
-  int  store(longlong nr, bool unsigned_val);
-  int  store_time_dec(MYSQL_TIME *ltime, uint dec);
-  my_decimal *val_decimal(my_decimal*);
-  bool eq_def(Field *field)
-  {
-    return (Field_str::eq_def(field) && decimals() == field->decimals());
-  }
-};
-
-class Field_date :public Field_temporal {
+class Field_date :public Field_temporal_with_date {
   void store_TIME(MYSQL_TIME *ltime);
 public:
   Field_date(uchar *ptr_arg, uchar *null_ptr_arg, uchar null_bit_arg,
-	     enum utype unireg_check_arg, const char *field_name_arg,
-	     CHARSET_INFO *cs)
-    :Field_temporal(ptr_arg, MAX_DATE_WIDTH, null_ptr_arg, null_bit_arg,
-                    unireg_check_arg, field_name_arg, cs) {}
+	     enum utype unireg_check_arg, const char *field_name_arg)
+    :Field_temporal_with_date(ptr_arg, MAX_DATE_WIDTH, null_ptr_arg, null_bit_arg,
+                              unireg_check_arg, field_name_arg) {}
   enum_field_types type() const { return MYSQL_TYPE_DATE;}
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONG_INT; }
   int reset(void) { ptr[0]=ptr[1]=ptr[2]=ptr[3]=0; return 0; }
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
-  uint decimals() const { return 0; }
   bool send_binary(Protocol *protocol);
   int cmp(const uchar *,const uchar *);
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const { return 4; }
   void sql_type(String &str) const;
-  bool zero_pack() const { return 1; }
   uchar *pack(uchar* to, const uchar *from,
               uint max_length __attribute__((unused)))
   {
@@ -1544,23 +1717,22 @@ public:
   {
     return unpack_int32(to, from, from_end);
   }
+  uint size_of() const { return sizeof(*this); }
 };
 
 
-class Field_newdate :public Field_temporal {
+class Field_newdate :public Field_temporal_with_date {
   void store_TIME(MYSQL_TIME *ltime);
 public:
   Field_newdate(uchar *ptr_arg, uchar *null_ptr_arg, uchar null_bit_arg,
-		enum utype unireg_check_arg, const char *field_name_arg,
-		CHARSET_INFO *cs)
-    :Field_temporal(ptr_arg, MAX_DATE_WIDTH, null_ptr_arg, null_bit_arg,
-                    unireg_check_arg, field_name_arg, cs)
+		enum utype unireg_check_arg, const char *field_name_arg)
+    :Field_temporal_with_date(ptr_arg, MAX_DATE_WIDTH, null_ptr_arg, null_bit_arg,
+                              unireg_check_arg, field_name_arg)
     {}
   enum_field_types type() const { return MYSQL_TYPE_DATE;}
   enum_field_types real_type() const { return MYSQL_TYPE_NEWDATE; }
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_UINT24; }
   int reset(void) { ptr[0]=ptr[1]=ptr[2]=0; return 0; }
-  uint decimals() const { return 0; }
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -1569,19 +1741,22 @@ public:
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const { return 3; }
   void sql_type(String &str) const;
-  bool zero_pack() const { return 1; }
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  uint size_of() const { return sizeof(*this); }
 };
 
 
 class Field_time :public Field_temporal {
-  void store_TIME(MYSQL_TIME *ltime);
+protected:
+  virtual void store_TIME(MYSQL_TIME *ltime);
+  int store_TIME_with_warning(MYSQL_TIME *ltime, const ErrConv *str,
+                              int was_cut, int have_smth_to_conv);
 public:
   Field_time(uchar *ptr_arg, uint length_arg, uchar *null_ptr_arg,
              uchar null_bit_arg, enum utype unireg_check_arg,
-             const char *field_name_arg, CHARSET_INFO *cs)
+             const char *field_name_arg)
     :Field_temporal(ptr_arg, length_arg, null_ptr_arg, null_bit_arg,
-                    unireg_check_arg, field_name_arg, cs)
+                    unireg_check_arg, field_name_arg)
     {}
   enum_field_types type() const { return MYSQL_TYPE_TIME;}
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_INT24; }
@@ -1589,7 +1764,7 @@ public:
   int store(const char *to,uint length,CHARSET_INFO *charset);
   int store(double nr);
   int store(longlong nr, bool unsigned_val);
-  uint decimals() const { return 0; }
+  int  store_decimal(const my_decimal *);
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -1599,55 +1774,122 @@ public:
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const { return 3; }
   void sql_type(String &str) const;
-  bool zero_pack() const { return 1; }
+  uint size_of() const { return sizeof(*this); }
 };
 
-class Field_time_hires :public Field_time {
+
+/**
+  Abstract class for:
+  - TIME(1..6)
+  - TIME(0..6) - MySQL56 version
+*/
+class Field_time_with_dec :public Field_time {
+protected:
   uint dec;
+public:
+  Field_time_with_dec(uchar *ptr_arg, uchar *null_ptr_arg, uchar null_bit_arg,
+                      enum utype unireg_check_arg, const char *field_name_arg,
+                      uint dec_arg)
+    :Field_time(ptr_arg, MIN_TIME_WIDTH + dec_arg + test(dec_arg), null_ptr_arg,
+                null_bit_arg, unireg_check_arg, field_name_arg),
+     dec(dec_arg)
+  {
+    DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  }
+  uint decimals() const { return dec; }
+  enum ha_base_keytype key_type() const { return HA_KEYTYPE_BINARY; }
+  longlong val_int(void);
+  double val_real(void);
+  void make_field(Send_field *);
+};
+
+
+/**
+  TIME(1..6)
+*/
+class Field_time_hires :public Field_time_with_dec {
   longlong zero_point;
   void store_TIME(MYSQL_TIME *ltime);
 public:
   Field_time_hires(uchar *ptr_arg, uchar *null_ptr_arg, uchar null_bit_arg,
              enum utype unireg_check_arg, const char *field_name_arg,
-             uint dec_arg, CHARSET_INFO *cs)
-    :Field_time(ptr_arg, MIN_TIME_WIDTH + dec_arg + 1, null_ptr_arg,
-                null_bit_arg, unireg_check_arg, field_name_arg, cs),
-     dec(dec_arg)
+             uint dec_arg)
+    :Field_time_with_dec(ptr_arg, null_ptr_arg,
+                         null_bit_arg, unireg_check_arg, field_name_arg,
+                         dec_arg)
   {
     DBUG_ASSERT(dec);
-    DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
     zero_point= sec_part_shift(
                    ((TIME_MAX_VALUE_SECONDS+1LL)*TIME_SECOND_PART_FACTOR), dec);
   }
-  enum ha_base_keytype key_type() const { return HA_KEYTYPE_BINARY; }
-  uint decimals() const { return dec; }
-  int store_decimal(const my_decimal *d);
-  longlong val_int(void);
-  double val_real(void);
-  String *val_str(String*,String *);
   int reset(void);
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
-  bool send_binary(Protocol *protocol);
   int cmp(const uchar *,const uchar *);
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const;
-  void sql_type(String &str) const;
-  void make_field(Send_field *);
   uint size_of() const { return sizeof(*this); }
 };
 
-class Field_datetime :public Field_temporal {
+
+/**
+  TIME(0..6) - MySQL56 version
+*/
+class Field_timef :public Field_time_with_dec {
+  void store_TIME(MYSQL_TIME *ltime);
+  int do_save_field_metadata(uchar *metadata_ptr)
+  {
+    *metadata_ptr= decimals();
+    return 1;
+  }
+public:
+  Field_timef(uchar *ptr_arg, uchar *null_ptr_arg, uchar null_bit_arg,
+             enum utype unireg_check_arg, const char *field_name_arg,
+             uint dec_arg)
+    :Field_time_with_dec(ptr_arg, null_ptr_arg,
+                         null_bit_arg, unireg_check_arg, field_name_arg,
+                         dec_arg)
+  {
+    DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
+  }
+  enum_field_types real_type() const { return MYSQL_TYPE_TIME2; }
+  enum_field_types binlog_type() const { return MYSQL_TYPE_TIME2; }
+  uint32 pack_length() const
+  {
+    return my_time_binary_length(dec);
+  }
+  uint row_pack_length() { return pack_length(); }
+  uint pack_length_from_metadata(uint field_metadata)
+  {
+    DBUG_ENTER("Field_timef::pack_length_from_metadata");
+    uint tmp= my_time_binary_length(field_metadata);
+    DBUG_RETURN(tmp);
+  }
+  void sort_string(uchar *to, uint length)
+  {
+    DBUG_ASSERT(length == Field_timef::pack_length());
+    memcpy(to, ptr, length);
+  }
+  int cmp(const uchar *a_ptr, const uchar *b_ptr)
+  {
+    return memcmp(a_ptr, b_ptr, pack_length());
+  }
+  int reset();
+  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  uint size_of() const { return sizeof(*this); }
+};
+
+
+class Field_datetime :public Field_temporal_with_date {
   void store_TIME(MYSQL_TIME *ltime);
 public:
   Field_datetime(uchar *ptr_arg, uint length_arg, uchar *null_ptr_arg,
                  uchar null_bit_arg, enum utype unireg_check_arg,
-                 const char *field_name_arg, CHARSET_INFO *cs)
-    :Field_temporal(ptr_arg, length_arg, null_ptr_arg, null_bit_arg,
-                    unireg_check_arg, field_name_arg, cs)
+                 const char *field_name_arg)
+    :Field_temporal_with_date(ptr_arg, length_arg, null_ptr_arg, null_bit_arg,
+                              unireg_check_arg, field_name_arg)
     {}
   enum_field_types type() const { return MYSQL_TYPE_DATETIME;}
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_ULONGLONG; }
-  uint decimals() const { return 0; }
   double val_real(void);
   longlong val_int(void);
   String *val_str(String*,String *);
@@ -1656,7 +1898,6 @@ public:
   void sort_string(uchar *buff,uint length);
   uint32 pack_length() const { return 8; }
   void sql_type(String &str) const;
-  bool zero_pack() const { return 1; }
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
   virtual int set_time();
   virtual void set_default()
@@ -1690,85 +1931,149 @@ public:
   {
     return unpack_int64(to, from, from_end);
   }
+  uint size_of() const { return sizeof(*this); }
 };
 
 
-class Field_datetime_hires :public Field_datetime {
-  void store_TIME(MYSQL_TIME *ltime);
+/**
+  Abstract class for:
+  - DATETIME(1..6)
+  - DATETIME(0..6) - MySQL56 version
+*/
+class Field_datetime_with_dec :public Field_datetime {
+protected:
   uint dec;
 public:
-  Field_datetime_hires(uchar *ptr_arg, uchar *null_ptr_arg,
-                       uchar null_bit_arg, enum utype unireg_check_arg,
-                       const char *field_name_arg, uint dec_arg,
-                       CHARSET_INFO *cs)
-    :Field_datetime(ptr_arg, MAX_DATETIME_WIDTH + dec_arg + 1,
+  Field_datetime_with_dec(uchar *ptr_arg, uchar *null_ptr_arg,
+                          uchar null_bit_arg, enum utype unireg_check_arg,
+                          const char *field_name_arg, uint dec_arg)
+    :Field_datetime(ptr_arg, MAX_DATETIME_WIDTH + dec_arg + test(dec_arg),
                     null_ptr_arg, null_bit_arg, unireg_check_arg,
-                    field_name_arg, cs), dec(dec_arg)
+                    field_name_arg), dec(dec_arg)
   {
-    DBUG_ASSERT(dec);
     DBUG_ASSERT(dec <= TIME_SECOND_PART_DIGITS);
   }
-  enum ha_base_keytype key_type() const { return HA_KEYTYPE_BINARY; }
   uint decimals() const { return dec; }
+  enum ha_base_keytype key_type() const { return HA_KEYTYPE_BINARY; }
   void make_field(Send_field *field);
-  int store_decimal(const my_decimal *d);
-  double val_real(void);
-  longlong val_int(void);
-  String *val_str(String*,String *);
   bool send_binary(Protocol *protocol);
-  int cmp(const uchar *,const uchar *);
-  void sort_string(uchar *buff,uint length);
-  uint32 pack_length() const;
-  void sql_type(String &str) const;
-  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
   uchar *pack(uchar *to, const uchar *from, uint max_length)
   { return Field::pack(to, from, max_length); }
   const uchar *unpack(uchar* to, const uchar *from, const uchar *from_end,
                       uint param_data)
   { return Field::unpack(to, from, from_end, param_data); }
+  void sort_string(uchar *to, uint length)
+  {
+    DBUG_ASSERT(length == pack_length());
+    memcpy(to, ptr, length);
+  }
+  double val_real(void);
+  longlong val_int(void);
+  String *val_str(String*,String *);
+};
+
+
+/**
+  DATETIME(1..6)
+*/
+class Field_datetime_hires :public Field_datetime_with_dec {
+  void store_TIME(MYSQL_TIME *ltime);
+public:
+  Field_datetime_hires(uchar *ptr_arg, uchar *null_ptr_arg,
+                       uchar null_bit_arg, enum utype unireg_check_arg,
+                       const char *field_name_arg, uint dec_arg)
+    :Field_datetime_with_dec(ptr_arg, null_ptr_arg, null_bit_arg,
+                             unireg_check_arg, field_name_arg, dec_arg)
+  {
+    DBUG_ASSERT(dec);
+  }
+  int cmp(const uchar *,const uchar *);
+  uint32 pack_length() const;
+  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
   uint size_of() const { return sizeof(*this); }
 };
 
+
+/**
+  DATETIME(0..6) - MySQL56 version
+*/
+class Field_datetimef :public Field_datetime_with_dec {
+  void store_TIME(MYSQL_TIME *ltime);
+  int do_save_field_metadata(uchar *metadata_ptr)
+  {
+    *metadata_ptr= decimals();
+    return 1;
+  }
+public:
+  Field_datetimef(uchar *ptr_arg, uchar *null_ptr_arg,
+                  uchar null_bit_arg, enum utype unireg_check_arg,
+                  const char *field_name_arg, uint dec_arg)
+    :Field_datetime_with_dec(ptr_arg, null_ptr_arg, null_bit_arg,
+                             unireg_check_arg, field_name_arg, dec_arg)
+  {}
+  enum_field_types real_type() const { return MYSQL_TYPE_DATETIME2; }
+  enum_field_types binlog_type() const { return MYSQL_TYPE_DATETIME2; }
+  uint32 pack_length() const
+  {
+    return my_datetime_binary_length(dec);
+  }
+  uint row_pack_length() { return pack_length(); }
+  uint pack_length_from_metadata(uint field_metadata)
+  {
+    DBUG_ENTER("Field_datetimef::pack_length_from_metadata");
+    uint tmp= my_datetime_binary_length(field_metadata);
+    DBUG_RETURN(tmp);
+  }
+  int cmp(const uchar *a_ptr, const uchar *b_ptr)
+  {
+    return memcmp(a_ptr, b_ptr, pack_length());
+  }
+  int reset();
+  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  uint size_of() const { return sizeof(*this); }
+};
+
+
 static inline Field_timestamp *
 new_Field_timestamp(uchar *ptr, uchar *null_ptr, uchar null_bit,
                     enum Field::utype unireg_check, const char *field_name,
-                    TABLE_SHARE *share, uint dec, CHARSET_INFO *cs)
+                    TABLE_SHARE *share, uint dec)
 {
   if (dec==0)
     return new Field_timestamp(ptr, MAX_DATETIME_WIDTH, null_ptr, null_bit,
-                                unireg_check, field_name, share, cs);
+                                unireg_check, field_name, share);
   if (dec == NOT_FIXED_DEC)
     dec= MAX_DATETIME_PRECISION;
   return new Field_timestamp_hires(ptr, null_ptr, null_bit, unireg_check,
-                                   field_name, share, dec, cs);
+                                   field_name, share, dec);
 }
 
 static inline Field_time *
 new_Field_time(uchar *ptr, uchar *null_ptr, uchar null_bit,
                enum Field::utype unireg_check, const char *field_name,
-               uint dec, CHARSET_INFO *cs)
+               uint dec)
 {
   if (dec == 0)
     return new Field_time(ptr, MIN_TIME_WIDTH, null_ptr, null_bit,
-                          unireg_check, field_name, cs);
+                          unireg_check, field_name);
   if (dec == NOT_FIXED_DEC)
     dec= MAX_DATETIME_PRECISION;
   return new Field_time_hires(ptr, null_ptr, null_bit,
-                                  unireg_check, field_name, dec, cs);
+                                  unireg_check, field_name, dec);
 }
 
 static inline Field_datetime *
 new_Field_datetime(uchar *ptr, uchar *null_ptr, uchar null_bit,
                    enum Field::utype unireg_check,
-                   const char *field_name, uint dec, CHARSET_INFO *cs)
+                   const char *field_name, uint dec)
 {
   if (dec == 0)
     return new Field_datetime(ptr, MAX_DATETIME_WIDTH, null_ptr, null_bit,
-                              unireg_check, field_name, cs);
+                              unireg_check, field_name);
   if (dec == NOT_FIXED_DEC)
     dec= MAX_DATETIME_PRECISION;
   return new Field_datetime_hires(ptr, null_ptr, null_bit,
-                                  unireg_check, field_name, dec, cs);
+                                  unireg_check, field_name, dec);
 }
 
 class Field_string :public Field_longstr {
@@ -1795,7 +2100,6 @@ public:
             orig_table->s->frm_version < FRM_VER_TRUE_VARCHAR ?
 	    MYSQL_TYPE_VAR_STRING : MYSQL_TYPE_STRING);
   }
-  bool match_collation_to_optimize_range() const { return TRUE; }
   enum ha_base_keytype key_type() const
     { return binary() ? HA_KEYTYPE_BINARY : HA_KEYTYPE_TEXT; }
   bool zero_pack() const { return 0; }
@@ -1876,7 +2180,6 @@ public:
   }
 
   enum_field_types type() const { return MYSQL_TYPE_VARCHAR; }
-  bool match_collation_to_optimize_range() const { return TRUE; }
   enum ha_base_keytype key_type() const;
   uint row_pack_length() { return field_length; }
   bool zero_pack() const { return 0; }
@@ -1972,7 +2275,6 @@ public:
     :Field_longstr((uchar*) 0, 0, (uchar*) "", 0, NONE, "temp", system_charset_info),
     packlength(packlength_arg) {}
   enum_field_types type() const { return MYSQL_TYPE_BLOB;}
-  bool match_collation_to_optimize_range() const { return TRUE; }
   enum ha_base_keytype key_type() const
     { return binary() ? HA_KEYTYPE_VARBINARY2 : HA_KEYTYPE_VARTEXT2; }
   int  store(const char *to,uint length,CHARSET_INFO *charset);
@@ -2118,7 +2420,7 @@ public:
   { geom_type= geom_type_arg; }
   enum ha_base_keytype key_type() const { return HA_KEYTYPE_VARBINARY2; }
   enum_field_types type() const { return MYSQL_TYPE_GEOMETRY; }
-  bool match_collation_to_optimize_range() const { return FALSE; }
+  bool match_collation_to_optimize_range() const { return false; }
   void sql_type(String &str) const;
   int  store(const char *to, uint length, CHARSET_INFO *charset);
   int  store(double nr);
@@ -2156,7 +2458,6 @@ public:
   }
   Field *new_field(MEM_ROOT *root, TABLE *new_table, bool keep_type);
   enum_field_types type() const { return MYSQL_TYPE_STRING; }
-  bool match_collation_to_optimize_range() const { return FALSE; }
   enum Item_result cmp_type () const { return INT_RESULT; }
   enum ha_base_keytype key_type() const;
   int  store(const char *to,uint length,CHARSET_INFO *charset);
@@ -2309,7 +2610,10 @@ public:
   {
     store(*((longlong *)val), TRUE);
   }
-  double pos_in_interval(Field *min, Field *max);
+  double pos_in_interval(Field *min, Field *max)
+  {
+    return pos_in_interval_val_real(min, max);
+  }
   void get_image(uchar *buff, uint length, CHARSET_INFO *cs)
   { get_key_image(buff, length, itRAW); }   
   void set_image(const uchar *buff,uint length, CHARSET_INFO *cs)
diff --git a/sql/item.cc b/sql/item.cc
index a5d5cf61496..f9def19bf6a 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -651,9 +651,12 @@ Item_result Item::cmp_type() const
   case MYSQL_TYPE_GEOMETRY:
                            return STRING_RESULT;
   case MYSQL_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_TIMESTAMP2:
   case MYSQL_TYPE_DATE:
   case MYSQL_TYPE_TIME:
+  case MYSQL_TYPE_TIME2:
   case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_DATETIME2:
   case MYSQL_TYPE_NEWDATE:
                            return TIME_RESULT;
   };
@@ -3218,11 +3221,7 @@ void Item_param::set_time(MYSQL_TIME *tm, timestamp_type time_type,
   value.time= *tm;
   value.time.time_type= time_type;
 
-  if (value.time.year > 9999 || value.time.month > 12 ||
-      value.time.day > 31 ||
-      (time_type != MYSQL_TIMESTAMP_TIME && value.time.hour > 23) ||
-      value.time.minute > 59 || value.time.second > 59 ||
-      value.time.second_part > TIME_MAX_SECOND_PART)
+  if (check_datetime_range(&value.time))
   {
     ErrConvTime str(&value.time);
     make_truncated_value_warning(current_thd, MYSQL_ERROR::WARN_LEVEL_WARN,
@@ -5678,19 +5677,17 @@ Field *Item::tmp_table_field_from_field_type(TABLE *table, bool fixed_length)
     break;
   case MYSQL_TYPE_NEWDATE:
   case MYSQL_TYPE_DATE:
-    field= new Field_newdate(0, null_ptr, 0, Field::NONE, name, &my_charset_bin);
+    field= new Field_newdate(0, null_ptr, 0, Field::NONE, name);
     break;
   case MYSQL_TYPE_TIME:
-    field= new_Field_time(0, null_ptr, 0, Field::NONE, name,
-                              decimals, &my_charset_bin);
+    field= new_Field_time(0, null_ptr, 0, Field::NONE, name, decimals);
     break;
   case MYSQL_TYPE_TIMESTAMP:
     field= new_Field_timestamp(0, null_ptr, 0,
-                               Field::NONE, name, 0, decimals, &my_charset_bin);
+                               Field::NONE, name, 0, decimals);
     break;
   case MYSQL_TYPE_DATETIME:
-    field= new_Field_datetime(0, null_ptr, 0, Field::NONE, name,
-                              decimals, &my_charset_bin);
+    field= new_Field_datetime(0, null_ptr, 0, Field::NONE, name, decimals);
     break;
   case MYSQL_TYPE_YEAR:
     field= new Field_year((uchar*) 0, max_length, null_ptr, 0, Field::NONE,
diff --git a/sql/item_strfunc.cc b/sql/item_strfunc.cc
index 5bf53784c26..132515733d6 100644
--- a/sql/item_strfunc.cc
+++ b/sql/item_strfunc.cc
@@ -3972,7 +3972,9 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
         type= DYN_COL_NULL;
         break;
       case MYSQL_TYPE_TIMESTAMP:
+      case MYSQL_TYPE_TIMESTAMP2:
       case MYSQL_TYPE_DATETIME:
+      case MYSQL_TYPE_DATETIME2:
         type= DYN_COL_DATETIME;
 	break;
       case MYSQL_TYPE_DATE:
@@ -3980,6 +3982,7 @@ bool Item_func_dyncol_create::prepare_arguments(bool force_names_arg)
         type= DYN_COL_DATE;
         break;
       case MYSQL_TYPE_TIME:
+      case MYSQL_TYPE_TIME2:
         type= DYN_COL_TIME;
         break;
       case MYSQL_TYPE_VARCHAR:
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index d441653c82b..92fff032095 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -1307,16 +1307,16 @@ Field *Item_sum_hybrid::create_tmp_field(bool group, TABLE *table,
   switch (args[0]->field_type()) {
   case MYSQL_TYPE_DATE:
     field= new Field_newdate(0, maybe_null ? (uchar*)"" : 0, 0, Field::NONE,
-                             name, collation.collation);
+                             name);
     break;
   case MYSQL_TYPE_TIME:
     field= new_Field_time(0, maybe_null ? (uchar*)"" : 0, 0, Field::NONE,
-                          name, decimals, collation.collation);
+                          name, decimals);
     break;
   case MYSQL_TYPE_TIMESTAMP:
   case MYSQL_TYPE_DATETIME:
     field= new_Field_datetime(0, maybe_null ? (uchar*)"" : 0, 0, Field::NONE,
-                              name, decimals, collation.collation);
+                              name, decimals);
     break;
   default:
     return Item_sum::create_tmp_field(group, table, convert_blob_length);
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index f43a4e79431..84510e0f112 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -1574,7 +1574,7 @@ static void set_sec_part(ulong sec_part, MYSQL_TIME *ltime, Item *item)
   {
     ltime->second_part= sec_part;
     if (item->decimals < TIME_SECOND_PART_DIGITS)
-      ltime->second_part= sec_part_truncate(ltime->second_part, item->decimals);
+      my_time_trunc(ltime, item->decimals);
   }
 }
 
@@ -2411,7 +2411,7 @@ bool Item_time_typecast::get_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
   if (get_arg0_time(ltime))
     return 1;
   if (decimals < TIME_SECOND_PART_DIGITS)
-    ltime->second_part= sec_part_truncate(ltime->second_part, decimals);
+    my_time_trunc(ltime, decimals);
   /*
     MYSQL_TIMESTAMP_TIME value can have non-zero day part,
     which we should not lose.
@@ -2450,8 +2450,7 @@ bool Item_datetime_typecast::get_date(MYSQL_TIME *ltime, ulonglong fuzzy_date)
     return 1;
 
   if (decimals < TIME_SECOND_PART_DIGITS)
-    ltime->second_part= sec_part_truncate(ltime->second_part, decimals);
-
+    my_time_trunc(ltime, decimals);
 
   /*
     ltime is valid MYSQL_TYPE_TIME (according to fuzzy_date).
diff --git a/sql/log_event.cc b/sql/log_event.cc
index 0782e40981e..6f56d1d53b9 100644
--- a/sql/log_event.cc
+++ b/sql/log_event.cc
@@ -48,6 +48,7 @@
 #include <my_dir.h>
 #include "sql_show.h"    // append_identifier
 #include <strfunc.h>
+#include "compat56.h"
 
 #endif /* MYSQL_CLIENT */
 
@@ -2128,6 +2129,17 @@ log_event_print_value(IO_CACHE *file, const uchar *ptr,
       return 4;
     }
 
+  case MYSQL_TYPE_TIMESTAMP2:
+    {
+      char buf[MAX_DATE_STRING_REP_LENGTH];
+      struct timeval tm;
+      my_timestamp_from_binary(&tm, ptr, meta);
+      int buflen= my_timeval_to_str(&tm, buf, meta);
+      my_b_write(file, buf, buflen);
+      my_snprintf(typestr, typestr_length, "TIMESTAMP(%d)", meta);
+      return my_timestamp_binary_length(meta);
+    }
+
   case MYSQL_TYPE_DATETIME:
     {
       ulong d, t;
@@ -2142,15 +2154,41 @@ log_event_print_value(IO_CACHE *file, const uchar *ptr,
       return 8;
     }
 
+  case MYSQL_TYPE_DATETIME2:
+    {
+      char buf[MAX_DATE_STRING_REP_LENGTH];
+      MYSQL_TIME ltime;
+      longlong packed= my_datetime_packed_from_binary(ptr, meta);
+      TIME_from_longlong_datetime_packed(&ltime, packed);
+      int buflen= my_datetime_to_str(&ltime, buf, meta);
+      my_b_write_quoted(file, (uchar *) buf, buflen);
+      my_snprintf(typestr, typestr_length, "DATETIME(%d)", meta);
+      return my_datetime_binary_length(meta);
+    }
+
   case MYSQL_TYPE_TIME:
     {
-      uint32 i32= uint3korr(ptr);
-      my_b_printf(file, "'%02d:%02d:%02d'",
-                  i32 / 10000, (i32 % 10000) / 100, i32 % 100);
+      int32 tmp= sint3korr(ptr);
+      int32 i32= tmp >= 0 ? tmp : - tmp;
+      const char *sign= tmp < 0 ? "-" : "";
+      my_b_printf(file, "'%s%02d:%02d:%02d'",
+                  sign, i32 / 10000, (i32 % 10000) / 100, i32 % 100, i32);
       my_snprintf(typestr,  typestr_length, "TIME");
       return 3;
     }
-    
+
+  case MYSQL_TYPE_TIME2:
+    {
+      char buf[MAX_DATE_STRING_REP_LENGTH];
+      MYSQL_TIME ltime;
+      longlong packed= my_time_packed_from_binary(ptr, meta);
+      TIME_from_longlong_time_packed(&ltime, packed);
+      int buflen= my_time_to_str(&ltime, buf, meta);
+      my_b_write_quoted(file, (uchar *) buf, buflen);
+      my_snprintf(typestr, typestr_length, "TIME(%d)", meta);
+      return my_time_binary_length(meta);
+    }
+
   case MYSQL_TYPE_NEWDATE:
     {
       uint32 tmp= uint3korr(ptr);
@@ -9860,7 +9898,7 @@ Table_map_log_event::Table_map_log_event(THD *thd, TABLE *tbl, ulong tid,
   {
     m_coltype= reinterpret_cast<uchar*>(m_memory);
     for (unsigned int i= 0 ; i < m_table->s->fields ; ++i)
-      m_coltype[i]= m_table->field[i]->type();
+      m_coltype[i]= m_table->field[i]->binlog_type();
   }
 
   /*
diff --git a/sql/opt_range.cc b/sql/opt_range.cc
index 659e63caf6c..e8213ea8dd6 100644
--- a/sql/opt_range.cc
+++ b/sql/opt_range.cc
@@ -8014,10 +8014,10 @@ get_mm_leaf(RANGE_OPT_PARAM *param, COND *conf_func, Field *field,
 
   */
   if (field->result_type() == STRING_RESULT &&
-      ((Field_str*) field)->match_collation_to_optimize_range() &&
+      field->match_collation_to_optimize_range() &&
       value->result_type() == STRING_RESULT &&
       key_part->image_type == Field::itRAW &&
-      ((Field_str*)field)->charset() != conf_func->compare_collation() &&
+      field->charset() != conf_func->compare_collation() &&
       !(conf_func->compare_collation()->state & MY_CS_BINSORT &&
         (type == Item_func::EQUAL_FUNC || type == Item_func::EQ_FUNC)))
     goto end;
@@ -12885,7 +12885,7 @@ check_group_min_max_predicates(Item *cond, Item_field *min_max_arg_item,
             */
             ((args[1]->result_type() == STRING_RESULT &&
               image_type == Field::itRAW &&
-              ((Field_str*) min_max_arg_item->field)->charset() !=
+              min_max_arg_item->field->charset() !=
               pred->compare_collation())
              ||
              /*
diff --git a/sql/opt_table_elimination.cc b/sql/opt_table_elimination.cc
index 33164c1ed12..1542efa7415 100644
--- a/sql/opt_table_elimination.cc
+++ b/sql/opt_table_elimination.cc
@@ -1482,7 +1482,7 @@ void check_equality(Dep_analysis_context *ctx, Dep_module_expr **eq_mod,
           collation of the operation differ from the field collation.
         */
         if (field->cmp_type() == STRING_RESULT &&
-            ((Field_str*)field)->charset() != cond->compare_collation())
+            field->charset() != cond->compare_collation())
           return;
       }
     }
diff --git a/sql/protocol.cc b/sql/protocol.cc
index f6e9e9e62e1..1e257b5a282 100644
--- a/sql/protocol.cc
+++ b/sql/protocol.cc
@@ -1427,7 +1427,7 @@ bool Protocol_binary::store(MYSQL_TIME *tm, int decimals)
   DBUG_ASSERT(decimals == AUTO_SEC_PART_DIGITS ||
               (decimals >= 0 && decimals <= TIME_SECOND_PART_DIGITS));
   if (decimals != AUTO_SEC_PART_DIGITS)
-    tm->second_part= sec_part_truncate(tm->second_part, decimals);
+    my_time_trunc(tm, decimals);
   int4store(pos+7, tm->second_part);
   if (tm->second_part)
     length=11;
@@ -1469,7 +1469,7 @@ bool Protocol_binary::store_time(MYSQL_TIME *tm, int decimals)
   DBUG_ASSERT(decimals == AUTO_SEC_PART_DIGITS ||
               (decimals >= 0 && decimals <= TIME_SECOND_PART_DIGITS));
   if (decimals != AUTO_SEC_PART_DIGITS)
-    tm->second_part= sec_part_truncate(tm->second_part, decimals);
+    my_time_trunc(tm, decimals);
   int4store(pos+8, tm->second_part);
   if (tm->second_part)
     length=12;
diff --git a/sql/rpl_utility.cc b/sql/rpl_utility.cc
index 6bbe998a624..ac8a8fe356b 100644
--- a/sql/rpl_utility.cc
+++ b/sql/rpl_utility.cc
@@ -109,12 +109,15 @@ max_display_length_for_field(enum_field_types sql_type, unsigned int metadata)
 
   case MYSQL_TYPE_DATE:
   case MYSQL_TYPE_TIME:
+  case MYSQL_TYPE_TIME2:
     return 3;
 
   case MYSQL_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_TIMESTAMP2:
     return 4;
 
   case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_DATETIME2:
     return 8;
 
   case MYSQL_TYPE_BIT:
@@ -262,12 +265,21 @@ uint32 table_def::calc_field_size(uint col, uchar *master_data) const
   case MYSQL_TYPE_TIME:
     length= 3;
     break;
+  case MYSQL_TYPE_TIME2:
+    length= my_time_binary_length(m_field_metadata[col]);
+    break;
   case MYSQL_TYPE_TIMESTAMP:
     length= 4;
     break;
+  case MYSQL_TYPE_TIMESTAMP2:
+    length= my_timestamp_binary_length(m_field_metadata[col]);
+    break;
   case MYSQL_TYPE_DATETIME:
     length= 8;
     break;
+  case MYSQL_TYPE_DATETIME2:
+    length= my_datetime_binary_length(m_field_metadata[col]);
+    break;
   case MYSQL_TYPE_BIT:
   {
     /*
@@ -376,6 +388,7 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
     break;
 
   case MYSQL_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_TIMESTAMP2:
     str->set_ascii(STRING_WITH_LEN("timestamp"));
     break;
 
@@ -393,10 +406,12 @@ void show_sql_type(enum_field_types type, uint16 metadata, String *str, CHARSET_
     break;
 
   case MYSQL_TYPE_TIME:
+  case MYSQL_TYPE_TIME2:
     str->set_ascii(STRING_WITH_LEN("time"));
     break;
 
   case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_DATETIME2:
     str->set_ascii(STRING_WITH_LEN("datetime"));
     break;
 
@@ -615,6 +630,23 @@ can_convert_field_to(Field *field,
     else
       DBUG_RETURN(false);
   }
+  else if (metadata == 0 &&
+           ((field->real_type() == MYSQL_TYPE_TIMESTAMP2 &&
+             source_type == MYSQL_TYPE_TIMESTAMP) ||
+            (field->real_type() == MYSQL_TYPE_TIME2 &&
+             source_type == MYSQL_TYPE_TIME) ||
+            (field->real_type() == MYSQL_TYPE_DATETIME2 &&
+             source_type == MYSQL_TYPE_DATETIME)))
+  {
+    /*
+      TS-TODO: conversion from FSP1>FSP2.
+      Can do non-lossy conversion
+      from old TIME, TIMESTAMP, DATETIME
+      to MySQL56 TIME(0), TIMESTAMP(0), DATETIME(0).
+    */
+    *order_var= -1;
+    DBUG_RETURN(true);
+  }
   else if (!slave_type_conversions_options)
     DBUG_RETURN(false);
 
@@ -739,6 +771,9 @@ can_convert_field_to(Field *field,
   case MYSQL_TYPE_NULL:
   case MYSQL_TYPE_ENUM:
   case MYSQL_TYPE_SET:
+  case MYSQL_TYPE_TIMESTAMP2:
+  case MYSQL_TYPE_DATETIME2:
+  case MYSQL_TYPE_TIME2:
     DBUG_RETURN(false);
   }
   DBUG_RETURN(false);                                 // To keep GCC happy
@@ -939,7 +974,7 @@ TABLE *table_def::create_conversion_table(THD *thd, Relay_log_info *rli, TABLE *
 
     DBUG_PRINT("debug", ("sql_type: %d, target_field: '%s', max_length: %d, decimals: %d,"
                          " maybe_null: %d, unsigned_flag: %d, pack_length: %u",
-                         type(col), target_table->field[col]->field_name,
+                         binlog_type(col), target_table->field[col]->field_name,
                          max_length, decimals, TRUE, FALSE, pack_length));
     field_def->init_for_tmp_table(type(col),
                                   max_length,
@@ -993,7 +1028,7 @@ table_def::table_def(unsigned char *types, ulong size,
     int index= 0;
     for (unsigned int i= 0; i < m_size; i++)
     {
-      switch (m_type[i]) {
+      switch (binlog_type(i)) {
       case MYSQL_TYPE_TINY_BLOB:
       case MYSQL_TYPE_BLOB:
       case MYSQL_TYPE_MEDIUM_BLOB:
@@ -1042,6 +1077,11 @@ table_def::table_def(unsigned char *types, ulong size,
         m_field_metadata[i]= x;
         break;
       }
+      case MYSQL_TYPE_TIME2:
+      case MYSQL_TYPE_DATETIME2:
+      case MYSQL_TYPE_TIMESTAMP2:
+        m_field_metadata[i]= field_metadata[index++];
+        break;
       default:
         m_field_metadata[i]= 0;
         break;
diff --git a/sql/rpl_utility.h b/sql/rpl_utility.h
index 79f4517c492..b08721aa8c2 100644
--- a/sql/rpl_utility.h
+++ b/sql/rpl_utility.h
@@ -65,6 +65,14 @@ public:
   ulong size() const { return m_size; }
 
 
+  /**
+    Returns internal binlog type code for one field,
+    without translation to real types.
+  */
+  enum_field_types binlog_type(ulong index) const
+  {
+    return static_cast<enum_field_types>(m_type[index]);
+  }
   /*
     Return a representation of the type data for one field.
 
@@ -82,7 +90,7 @@ public:
       either MYSQL_TYPE_STRING, MYSQL_TYPE_ENUM, or MYSQL_TYPE_SET, so
       we might need to modify the type to get the real type.
     */
-    enum_field_types source_type= static_cast<enum_field_types>(m_type[index]);
+    enum_field_types source_type= binlog_type(index);
     uint16 source_metadata= m_field_metadata[index];
     switch (source_type)
     {
diff --git a/sql/sql_partition.cc b/sql/sql_partition.cc
index 2bcefce1212..5bac6643772 100644
--- a/sql/sql_partition.cc
+++ b/sql/sql_partition.cc
@@ -1563,7 +1563,7 @@ bool field_is_partition_charset(Field *field)
       !(field->type() == MYSQL_TYPE_VARCHAR))
     return FALSE;
   {
-    CHARSET_INFO *cs= ((Field_str*)field)->charset();
+    CHARSET_INFO *cs= field->charset();
     if (!(field->type() == MYSQL_TYPE_STRING) ||
         !(cs->state & MY_CS_BINSORT))
       return TRUE;
@@ -1606,7 +1606,7 @@ bool check_part_func_fields(Field **ptr, bool ok_with_charsets)
     */
     if (field_is_partition_charset(field))
     {
-      CHARSET_INFO *cs= ((Field_str*)field)->charset();
+      CHARSET_INFO *cs= field->charset();
       if (!ok_with_charsets ||
           cs->mbmaxlen > 1 ||
           cs->strxfrm_multiply > 1)
@@ -2090,6 +2090,8 @@ static int check_part_field(enum_field_types sql_type,
     case MYSQL_TYPE_DATE:
     case MYSQL_TYPE_TIME:
     case MYSQL_TYPE_DATETIME:
+    case MYSQL_TYPE_TIME2:
+    case MYSQL_TYPE_DATETIME2:
       *result_type= STRING_RESULT;
       *need_cs_check= TRUE;
       return FALSE;
@@ -2102,6 +2104,7 @@ static int check_part_field(enum_field_types sql_type,
     case MYSQL_TYPE_NEWDECIMAL:
     case MYSQL_TYPE_DECIMAL:
     case MYSQL_TYPE_TIMESTAMP:
+    case MYSQL_TYPE_TIMESTAMP2:
     case MYSQL_TYPE_NULL:
     case MYSQL_TYPE_FLOAT:
     case MYSQL_TYPE_DOUBLE:
@@ -2974,7 +2977,7 @@ static void copy_to_part_field_buffers(Field **ptr,
     restore_ptr++;
     if (!field->maybe_null() || !field->is_null())
     {
-      CHARSET_INFO *cs= ((Field_str*)field)->charset();
+      CHARSET_INFO *cs= field->charset();
       uint max_len= field->pack_length();
       uint data_len= field->data_length();
       uchar *field_buf= *field_bufs;
diff --git a/sql/sql_prepare.cc b/sql/sql_prepare.cc
index 518b86ee301..51c3894ec3e 100644
--- a/sql/sql_prepare.cc
+++ b/sql/sql_prepare.cc
@@ -4409,7 +4409,7 @@ bool Protocol_local::store(const char *str, size_t length,
 bool Protocol_local::store(MYSQL_TIME *time, int decimals)
 {
   if (decimals != AUTO_SEC_PART_DIGITS)
-    time->second_part= sec_part_truncate(time->second_part, decimals);
+    my_time_trunc(time, decimals);
   return store_column(time, sizeof(MYSQL_TIME));
 }
 
@@ -4427,7 +4427,7 @@ bool Protocol_local::store_date(MYSQL_TIME *time)
 bool Protocol_local::store_time(MYSQL_TIME *time, int decimals)
 {
   if (decimals != AUTO_SEC_PART_DIGITS)
-    time->second_part= sec_part_truncate(time->second_part, decimals);
+    my_time_trunc(time, decimals);
   return store_column(time, sizeof(MYSQL_TIME));
 }
 
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index 806017b9ae5..93f8711fb7e 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -4307,7 +4307,7 @@ add_key_field(JOIN *join,
       {
         if ((*value)->cmp_type() != STRING_RESULT)
             return;
-        if (((Field_str*)field)->charset() != cond->compare_collation())
+        if (field->charset() != cond->compare_collation())
           return;
       }
     }
@@ -11857,7 +11857,7 @@ static bool check_simple_equality(Item *left_item, Item *right_item,
 
       if (field_item->cmp_type() == STRING_RESULT)
       {
-        CHARSET_INFO *cs= ((Field_str*) field_item->field)->charset();
+        CHARSET_INFO *cs= field_item->field->charset();
         if (!item)
         {
           Item_func_eq *eq_item;
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index f56eedf4c42..4b3997dec3c 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2790,6 +2790,8 @@ int prepare_create_field(Create_field *sql_field,
   case MYSQL_TYPE_NEWDATE:
   case MYSQL_TYPE_TIME:
   case MYSQL_TYPE_DATETIME:
+  case MYSQL_TYPE_TIME2:
+  case MYSQL_TYPE_DATETIME2:
   case MYSQL_TYPE_NULL:
     sql_field->pack_flag=f_settype((uint) sql_field->sql_type);
     break;
@@ -2808,6 +2810,7 @@ int prepare_create_field(Create_field *sql_field,
                           (sql_field->decimals << FIELDFLAG_DEC_SHIFT));
     break;
   case MYSQL_TYPE_TIMESTAMP:
+  case MYSQL_TYPE_TIMESTAMP2:
     /* fall-through */
   default:
     sql_field->pack_flag=(FIELDFLAG_NUMBER |
@@ -2893,7 +2896,7 @@ void promote_first_timestamp_column(List<Create_field> *column_definitions)
 
   while ((column_definition= it++) != NULL)
   {
-    if (column_definition->sql_type == MYSQL_TYPE_TIMESTAMP ||      // TIMESTAMP
+    if (is_timestamp_type(column_definition->sql_type) ||              // TIMESTAMP
         column_definition->unireg_check == Field::TIMESTAMP_OLD_FIELD) // Legacy
     {
       if ((column_definition->flags & NOT_NULL_FLAG) != 0 && // NOT NULL,
@@ -3854,7 +3857,7 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 
     if (thd->variables.sql_mode & MODE_NO_ZERO_DATE &&
         !sql_field->def &&
-        sql_field->sql_type == MYSQL_TYPE_TIMESTAMP &&
+        is_timestamp_type(sql_field->sql_type) &&
         (sql_field->flags & NOT_NULL_FLAG) &&
         (type == Field::NONE || type == Field::TIMESTAMP_UN_FIELD))
     {
@@ -5950,7 +5953,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
     */
     if ((def->sql_type == MYSQL_TYPE_DATE ||
          def->sql_type == MYSQL_TYPE_NEWDATE ||
-         def->sql_type == MYSQL_TYPE_DATETIME) &&
+         def->sql_type == MYSQL_TYPE_DATETIME ||
+         def->sql_type == MYSQL_TYPE_DATETIME2) &&
          !alter_info->datetime_field &&
          !(~def->flags & (NO_DEFAULT_VALUE_FLAG | NOT_NULL_FLAG)) &&
          thd->variables.sql_mode & MODE_NO_ZERO_DATE)
@@ -7582,6 +7586,7 @@ err:
         t_type= MYSQL_TIMESTAMP_DATE;
         break;
       case MYSQL_TYPE_DATETIME:
+      case MYSQL_TYPE_DATETIME2:
         f_val= "0000-00-00 00:00:00";
         t_type= MYSQL_TIMESTAMP_DATETIME;
         break;
diff --git a/sql/table.cc b/sql/table.cc
index 2f867eb7dc4..32a841ffd46 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -3281,7 +3281,7 @@ bool get_field(MEM_ROOT *mem, Field *field, String *res)
   }
   if (!(to= strmake_root(mem, str.ptr(), length)))
     length= 0;                                  // Safety fix
-  res->set(to, length, ((Field_str*)field)->charset());
+  res->set(to, length, field->charset());
   return 0;
 }