MDEV-13995 MAX(timestamp) returns a wrong result near DST changebb-10.4-mdev13648

22 files changed, 1625 insertions, 15 deletions

diff --git a/mysql-test/main/timezone2.result b/mysql-test/main/timezone2.result
index 6de62d7ea30..a58a0ee632f 100644
--- a/mysql-test/main/timezone2.result
+++ b/mysql-test/main/timezone2.result
@@ -353,5 +353,194 @@ Warning	1292	Truncated incorrect datetime value: '00:00:00'
 SET old_mode=DEFAULT;
 SET timestamp=DEFAULT;
 #
+# MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+#
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526)      /*summer time in Moscow*/);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526+3599) /*winter time in Moscow*/);
+SET time_zone='Europe/Moscow';
+SELECT a, UNIX_TIMESTAMP(a) FROM t1;
+a	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:26	1288477526
+2010-10-31 02:25:25	1288481125
+SELECT UNIX_TIMESTAMP(MAX(a)) AS a FROM t1;
+a
+1288481125
+CREATE TABLE t2 (a TIMESTAMP);
+INSERT INTO t2 SELECT MAX(a) AS a FROM t1;
+SELECT a, UNIX_TIMESTAMP(a) FROM t2;
+a	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:25	1288481125
+DROP TABLE t2;
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+CREATE TABLE t2 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526)      /*summer time in Moscow*/);
+INSERT INTO t2 VALUES (FROM_UNIXTIME(1288477526+3599) /*winter time in Moscow*/);
+SET time_zone='Europe/Moscow';
+SELECT UNIX_TIMESTAMP(t1.a), UNIX_TIMESTAMP(t2.a) FROM t1,t2;
+UNIX_TIMESTAMP(t1.a)	UNIX_TIMESTAMP(t2.a)
+1288477526	1288481125
+SELECT * FROM t1,t2 WHERE t1.a < t2.a;
+a	a
+2010-10-31 02:25:26	2010-10-31 02:25:25
+DROP TABLE t1,t2;
+BEGIN NOT ATOMIC
+DECLARE a,b TIMESTAMP;
+SET time_zone='+00:00';
+SET a=FROM_UNIXTIME(1288477526);
+SET b=FROM_UNIXTIME(1288481125);
+SELECT a < b;
+SET time_zone='Europe/Moscow';
+SELECT a < b;
+END;
+$$
+a < b
+1
+a < b
+1
+CREATE OR REPLACE FUNCTION f1(uts INT) RETURNS TIMESTAMP
+BEGIN
+DECLARE ts TIMESTAMP;
+DECLARE tz VARCHAR(64) DEFAULT @@time_zone;
+SET time_zone='+00:00';
+SET ts=FROM_UNIXTIME(uts);
+SET time_zone=tz;
+RETURN ts;
+END;
+$$
+SET time_zone='+00:00';
+SELECT f1(1288477526) < f1(1288481125);
+f1(1288477526) < f1(1288481125)
+1
+SET time_zone='Europe/Moscow';
+SELECT f1(1288477526) < f1(1288481125);
+f1(1288477526) < f1(1288481125)
+1
+DROP FUNCTION f1;
+CREATE TABLE t1 (a TIMESTAMP,b TIMESTAMP);
+SET time_zone='+00:00';
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526) /*summer time in Mowcow*/,
+FROM_UNIXTIME(1288481125) /*winter time in Moscow*/);
+SELECT *, LEAST(a,b) FROM t1;
+a	b	LEAST(a,b)
+2010-10-30 22:25:26	2010-10-30 23:25:25	2010-10-30 22:25:26
+SET time_zone='Europe/Moscow';
+SELECT *, LEAST(a,b) FROM t1;
+a	b	LEAST(a,b)
+2010-10-31 02:25:26	2010-10-31 02:25:25	2010-10-31 02:25:26
+SELECT UNIX_TIMESTAMP(a), UNIX_TIMESTAMP(b), UNIX_TIMESTAMP(LEAST(a,b)) FROM t1;
+UNIX_TIMESTAMP(a)	UNIX_TIMESTAMP(b)	UNIX_TIMESTAMP(LEAST(a,b))
+1288477526	1288481125	1288477526
+DROP TABLE t1;
+CREATE TABLE t1 (a TIMESTAMP,b TIMESTAMP,c TIMESTAMP);
+SET time_zone='+00:00';
+INSERT INTO t1 VALUES (
+FROM_UNIXTIME(1288477526) /*summer time in Moscow*/,
+FROM_UNIXTIME(1288481125) /*winter time in Moscow*/,
+FROM_UNIXTIME(1288481126) /*winter time in Moscow*/);
+SELECT b BETWEEN a AND c FROM t1;
+b BETWEEN a AND c
+1
+SET time_zone='Europe/Moscow';
+SELECT b BETWEEN a AND c FROM t1;
+b BETWEEN a AND c
+1
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526) /*summer time in Mowcow*/);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288481125) /*winter time in Moscow*/);
+SELECT a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+a	UNIX_TIMESTAMP(a)
+2010-10-30 22:25:26	1288477526
+2010-10-30 23:25:25	1288481125
+SELECT COALESCE(a) AS a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+a	UNIX_TIMESTAMP(a)
+2010-10-30 22:25:26	1288477526
+2010-10-30 23:25:25	1288481125
+SET time_zone='Europe/Moscow';
+SELECT a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+a	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:26	1288477526
+2010-10-31 02:25:25	1288481125
+SELECT COALESCE(a) AS a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+a	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:26	1288477526
+2010-10-31 02:25:25	1288481125
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526) /*summer time in Mowcow*/);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288481126) /*winter time in Moscow*/);
+SET time_zone='Europe/Moscow';
+SELECT a, UNIX_TIMESTAMP(a) FROM t1 GROUP BY a;
+a	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:26	1288477526
+2010-10-31 02:25:26	1288481126
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126));
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),CASE a WHEN b THEN 'eq' ELSE 'ne' END AS x FROM t1;
+UNIX_TIMESTAMP(a)	UNIX_TIMESTAMP(b)	x
+1288477526	1288481126	ne
+SET time_zone='Europe/Moscow';
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),CASE a WHEN b THEN 'eq' ELSE 'ne' END AS x FROM t1;
+UNIX_TIMESTAMP(a)	UNIX_TIMESTAMP(b)	x
+1288477526	1288481126	ne
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP,c TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126),FROM_UNIXTIME(1288481127));
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),a IN (b,c) AS x FROM t1;
+UNIX_TIMESTAMP(a)	UNIX_TIMESTAMP(b)	x
+1288477526	1288481126	0
+SET time_zone='Europe/Moscow';
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),a IN (b,c) AS x FROM t1;
+UNIX_TIMESTAMP(a)	UNIX_TIMESTAMP(b)	x
+1288477526	1288481126	0
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126));
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+a	b
+SET time_zone='Europe/Moscow';
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+a	b
+DROP TABLE t1;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1100000000),FROM_UNIXTIME(1200000000));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1100000001),FROM_UNIXTIME(1200000001));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1300000000),FROM_UNIXTIME(1400000000));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1300000001),FROM_UNIXTIME(1400000001));
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+a	b
+SET time_zone='Europe/Moscow';
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+a	b
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+a	b
+DROP TABLE t1;
+#
 # End of 10.4 tests
 #
diff --git a/mysql-test/main/timezone2.test b/mysql-test/main/timezone2.test
index 773b40ec86c..e0c6eb08108 100644
--- a/mysql-test/main/timezone2.test
+++ b/mysql-test/main/timezone2.test
@@ -325,5 +325,172 @@ SET old_mode=DEFAULT;
 SET timestamp=DEFAULT;
 
 --echo #
+--echo # MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+--echo #
+
+# MAX()
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526)      /*summer time in Moscow*/);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526+3599) /*winter time in Moscow*/);
+SET time_zone='Europe/Moscow';
+SELECT a, UNIX_TIMESTAMP(a) FROM t1;
+SELECT UNIX_TIMESTAMP(MAX(a)) AS a FROM t1;
+CREATE TABLE t2 (a TIMESTAMP);
+INSERT INTO t2 SELECT MAX(a) AS a FROM t1;
+SELECT a, UNIX_TIMESTAMP(a) FROM t2;
+DROP TABLE t2;
+DROP TABLE t1;
+
+
+# Comparison
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+CREATE TABLE t2 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526)      /*summer time in Moscow*/);
+INSERT INTO t2 VALUES (FROM_UNIXTIME(1288477526+3599) /*winter time in Moscow*/);
+SET time_zone='Europe/Moscow';
+SELECT UNIX_TIMESTAMP(t1.a), UNIX_TIMESTAMP(t2.a) FROM t1,t2;
+SELECT * FROM t1,t2 WHERE t1.a < t2.a;
+DROP TABLE t1,t2;
+
+
+# SP variable comparison
+DELIMITER $$;
+BEGIN NOT ATOMIC
+  DECLARE a,b TIMESTAMP;
+  SET time_zone='+00:00';
+  SET a=FROM_UNIXTIME(1288477526);
+  SET b=FROM_UNIXTIME(1288481125);
+  SELECT a < b;
+  SET time_zone='Europe/Moscow';
+  SELECT a < b;
+END;
+$$
+DELIMITER ;$$
+
+
+# SP function comparison
+DELIMITER $$;
+CREATE OR REPLACE FUNCTION f1(uts INT) RETURNS TIMESTAMP
+BEGIN
+  DECLARE ts TIMESTAMP;
+  DECLARE tz VARCHAR(64) DEFAULT @@time_zone;
+  SET time_zone='+00:00';
+  SET ts=FROM_UNIXTIME(uts);
+  SET time_zone=tz;
+  RETURN ts;
+END;
+$$
+DELIMITER ;$$
+SET time_zone='+00:00';
+SELECT f1(1288477526) < f1(1288481125);
+SET time_zone='Europe/Moscow';
+SELECT f1(1288477526) < f1(1288481125);
+DROP FUNCTION f1;
+
+
+# LEAST()
+CREATE TABLE t1 (a TIMESTAMP,b TIMESTAMP);
+SET time_zone='+00:00';
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526) /*summer time in Mowcow*/,
+                       FROM_UNIXTIME(1288481125) /*winter time in Moscow*/);
+SELECT *, LEAST(a,b) FROM t1;
+SET time_zone='Europe/Moscow';
+SELECT *, LEAST(a,b) FROM t1;
+SELECT UNIX_TIMESTAMP(a), UNIX_TIMESTAMP(b), UNIX_TIMESTAMP(LEAST(a,b)) FROM t1;
+DROP TABLE t1;
+
+
+# BETWEEN
+CREATE TABLE t1 (a TIMESTAMP,b TIMESTAMP,c TIMESTAMP);
+SET time_zone='+00:00';
+INSERT INTO t1 VALUES (
+  FROM_UNIXTIME(1288477526) /*summer time in Moscow*/,
+  FROM_UNIXTIME(1288481125) /*winter time in Moscow*/,
+  FROM_UNIXTIME(1288481126) /*winter time in Moscow*/);
+SELECT b BETWEEN a AND c FROM t1;
+SET time_zone='Europe/Moscow';
+SELECT b BETWEEN a AND c FROM t1;
+DROP TABLE t1;
+
+
+# ORDER BY
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526) /*summer time in Mowcow*/);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288481125) /*winter time in Moscow*/);
+SELECT a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+SELECT COALESCE(a) AS a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+SET time_zone='Europe/Moscow';
+SELECT a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+SELECT COALESCE(a) AS a, UNIX_TIMESTAMP(a) FROM t1 ORDER BY a;
+DROP TABLE t1;
+
+
+# GROUP BY
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526) /*summer time in Mowcow*/);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288481126) /*winter time in Moscow*/);
+SET time_zone='Europe/Moscow';
+SELECT a, UNIX_TIMESTAMP(a) FROM t1 GROUP BY a;
+DROP TABLE t1;
+
+
+# CASE
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126));
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),CASE a WHEN b THEN 'eq' ELSE 'ne' END AS x FROM t1;
+SET time_zone='Europe/Moscow';
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),CASE a WHEN b THEN 'eq' ELSE 'ne' END AS x FROM t1;
+DROP TABLE t1;
+
+
+# IN
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP,c TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126),FROM_UNIXTIME(1288481127));
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),a IN (b,c) AS x FROM t1;
+SET time_zone='Europe/Moscow';
+SELECT UNIX_TIMESTAMP(a),UNIX_TIMESTAMP(b),a IN (b,c) AS x FROM t1;
+DROP TABLE t1;
+
+# Comparison and IN in combination with a subquery (with one row)
+
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126));
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+
+SET time_zone='Europe/Moscow';
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+DROP TABLE t1;
+
+# Comparison and IN in combinarion with a subquery (with multiple rows)
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1100000000),FROM_UNIXTIME(1200000000));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1100000001),FROM_UNIXTIME(1200000001));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1288477526),FROM_UNIXTIME(1288481126));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1300000000),FROM_UNIXTIME(1400000000));
+INSERT INTO t1 VALUES (FROM_UNIXTIME(1300000001),FROM_UNIXTIME(1400000001));
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+
+SET time_zone='Europe/Moscow';
+SELECT * FROM t1 WHERE a = (SELECT MAX(b) FROM t1);
+SELECT * FROM t1 WHERE a = (SELECT MIN(b) FROM t1);
+SELECT * FROM t1 WHERE a IN ((SELECT MAX(b) FROM t1), (SELECT MIN(b) FROM t1));
+DROP TABLE t1;
+
+
+--echo #
 --echo # End of 10.4 tests
 --echo #
diff --git a/mysql-test/main/type_timestamp.result b/mysql-test/main/type_timestamp.result
index a708ef5ec34..97ab8744b8c 100644
--- a/mysql-test/main/type_timestamp.result
+++ b/mysql-test/main/type_timestamp.result
@@ -1033,5 +1033,25 @@ CREATE TABLE t1 (a TIMESTAMP);
 INSERT INTO t1 SELECT CAST(20010101 AS UNSIGNED);
 DROP TABLE t1;
 #
+# MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+#
+# Testing Item_func_rollup_const::val_native()
+# There is a bug in the below output (MDEV-16612)
+# Please remove this comment when MDEV-16612 is fixed and results are re-recorded
+CREATE TABLE t1 (id INT);
+INSERT INTO t1 VALUES (1),(2);
+BEGIN NOT ATOMIC
+DECLARE v TIMESTAMP DEFAULT '2001-01-01 10:20:30'; -- "v" will be wrapped into Item_func_rollup_const
+SELECT id, v AS v, COUNT(*) FROM t1 GROUP BY id,v WITH ROLLUP;
+END;
+$$
+id	v	COUNT(*)
+1	2001-01-01 10:20:30	1
+1	2001-01-01 10:20:30	1
+2	2001-01-01 10:20:30	1
+2	2001-01-01 10:20:30	1
+NULL	2001-01-01 10:20:30	2
+DROP TABLE t1;
+#
 # End of 10.4 tests
 #
diff --git a/mysql-test/main/type_timestamp.test b/mysql-test/main/type_timestamp.test
index eeb6e2f8213..7fb3f5d6cd2 100644
--- a/mysql-test/main/type_timestamp.test
+++ b/mysql-test/main/type_timestamp.test
@@ -626,5 +626,25 @@ INSERT INTO t1 SELECT CAST(20010101 AS UNSIGNED);
 DROP TABLE t1;
 
 --echo #
+--echo # MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+--echo #
+
+--echo # Testing Item_func_rollup_const::val_native()
+
+--echo # There is a bug in the below output (MDEV-16612)
+--echo # Please remove this comment when MDEV-16612 is fixed and results are re-recorded
+
+CREATE TABLE t1 (id INT);
+INSERT INTO t1 VALUES (1),(2);
+DELIMITER $$;
+BEGIN NOT ATOMIC
+  DECLARE v TIMESTAMP DEFAULT '2001-01-01 10:20:30'; -- "v" will be wrapped into Item_func_rollup_const
+  SELECT id, v AS v, COUNT(*) FROM t1 GROUP BY id,v WITH ROLLUP;
+END;
+$$
+DELIMITER ;$$
+DROP TABLE t1;
+
+--echo #
 --echo # End of 10.4 tests
 --echo #
diff --git a/sql/compat56.h b/sql/compat56.h
index bb5e2670f7d..ff887ebf1bb 100644
--- a/sql/compat56.h
+++ b/sql/compat56.h
@@ -19,6 +19,15 @@
 
 
 /** MySQL56 routines and macros **/
+
+/*
+  Buffer size for a native TIMESTAMP representation, for use with NativBuffer.
+  4 bytes for seconds
+  3 bytes for microseconds
+  1 byte for the trailing '\0' (class Native reserves extra 1 byte for '\0')
+*/
+#define STRING_BUFFER_TIMESTAMP_BINARY_SIZE    8 /* 4 + 3 + 1 */
+
 #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))
diff --git a/sql/field.cc b/sql/field.cc
index 6b38b38782c..7cb1ed404e6 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -5026,6 +5026,22 @@ my_time_t Field_timestamp::get_timestamp(const uchar *pos,
 }
 
 
+bool Field_timestamp::val_native(Native *to)
+{
+  int32 nr= sint4korr(ptr);
+  if (!nr)
+  {
+    to->length(0); // Zero datetime '0000-00-00 00:00:00'
+    return false;
+  }
+  if (to->alloc(4))
+    return true;
+  mi_int4store((uchar *) to->ptr(), nr);
+  to->length(4);
+  return false;
+}
+
+
 int Field_timestamp::store_TIME_with_warning(THD *thd, const Datetime *dt,
                                              const ErrConv *str, int was_cut)
 {
@@ -5413,6 +5429,18 @@ my_time_t Field_timestamp_hires::get_timestamp(const uchar *pos,
   return mi_uint4korr(pos);
 }
 
+
+bool Field_timestamp_hires::val_native(Native *to)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  struct timeval tm;
+  tm.tv_sec= mi_uint4korr(ptr);
+  tm.tv_usec= sec_part_unshift(read_bigendian(ptr+4, sec_part_bytes(dec)), dec);
+  return Timestamp_or_zero_datetime(Timestamp(tm), tm.tv_sec == 0).
+           to_native(to, dec);
+}
+
+
 double Field_timestamp_with_dec::val_real(void)
 {
   MYSQL_TIME ltime;
diff --git a/sql/field.h b/sql/field.h
index 6c0a981fc0a..6c585af3b02 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -785,6 +785,15 @@ public:
   virtual int  store_decimal(const my_decimal *d)=0;
   virtual int  store_time_dec(const MYSQL_TIME *ltime, uint dec);
   virtual int  store_timestamp(my_time_t timestamp, ulong sec_part);
+  /**
+    Store a value represented in native format
+  */
+  virtual int store_native(const Native &value)
+  {
+    DBUG_ASSERT(0);
+    reset();
+    return 0;
+  }
   int store_time(const MYSQL_TIME *ltime)
   { return store_time_dec(ltime, TIME_SECOND_PART_DIGITS); }
   int store(const char *to, size_t length, CHARSET_INFO *cs,
@@ -831,6 +840,11 @@ public:
      This trickery is used to decrease a number of malloc calls.
   */
   virtual String *val_str(String*,String *)=0;
+  virtual bool val_native(Native *to)
+  {
+    DBUG_ASSERT(!is_null());
+    return to->copy((const char *) ptr, pack_length());
+  }
   String *val_int_as_str(String *val_buffer, bool unsigned_flag);
   /*
     Return the field value as a LEX_CSTRING, without padding to full length
@@ -2780,6 +2794,16 @@ public:
     store_TIMESTAMP(Timestamp(ts, sec_part).round(decimals(), mode, &warn));
   }
   bool get_date(MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  int store_native(const Native &value)
+  {
+    Timestamp_or_zero_datetime tm(value);
+    if (tm.is_zero_datetime())
+      reset();
+    else
+      store_TIMESTAMP(tm);
+    return 0;
+  }
+  bool val_native(Native *to);
   uchar *pack(uchar *to, const uchar *from,
               uint max_length __attribute__((unused)))
   {
@@ -2859,6 +2883,7 @@ public:
   {
     DBUG_ASSERT(dec);
   }
+  bool val_native(Native *to);
   my_time_t get_timestamp(const uchar *pos, ulong *sec_part) const;
   int cmp(const uchar *,const uchar *);
   uint32 pack_length() const { return 4 + sec_part_bytes(dec); }
@@ -2909,6 +2934,16 @@ public:
   {
     return get_timestamp(ptr, sec_part);
   }
+  bool val_native(Native *to)
+  {
+    // Check if it's '0000-00-00 00:00:00' rather than a real timestamp
+    if (ptr[0] == 0 && ptr[1] == 0 && ptr[2] == 0 && ptr[3] == 0)
+    {
+      to->length(0);
+      return false;
+    }
+    return Field::val_native(to);
+  }
   uint size_of() const { return sizeof(*this); }
 };
 
diff --git a/sql/filesort.cc b/sql/filesort.cc
index d0cb77ac63d..20dfff615c8 100644
--- a/sql/filesort.cc
+++ b/sql/filesort.cc
@@ -1068,6 +1068,28 @@ Type_handler_temporal_result::make_sort_key(uchar *to, Item *item,
 
 
 void
+Type_handler_timestamp_common::make_sort_key(uchar *to, Item *item,
+                                             const SORT_FIELD_ATTR *sort_field,
+                                             Sort_param *param) const
+{
+  Timestamp::NativeBuffer native;
+  uint binlen= my_timestamp_binary_length(item->decimals);
+  if (item->val_native(current_thd, &native) || native.length() == 0)
+  {
+    // NULL or '0000-00-00 00:00:00'
+    bzero(to, item->maybe_null ? binlen + 1 : binlen);
+  }
+  else
+  {
+    DBUG_ASSERT(native.length() == binlen);
+    if (item->maybe_null)
+      *to++= 1;
+    memcpy((char *) to, native.ptr(), binlen);
+  }
+}
+
+
+void
 Type_handler::make_sort_key_longlong(uchar *to,
                                      bool maybe_null,
                                      bool null_value,
@@ -1874,6 +1896,15 @@ Type_handler_temporal_result::sortlength(THD *thd,
 
 
 void
+Type_handler_timestamp_common::sortlength(THD *thd,
+                                          const Type_std_attributes *item,
+                                          SORT_FIELD_ATTR *sortorder) const
+{
+  sortorder->length= my_timestamp_binary_length(item->decimals);
+}
+
+
+void
 Type_handler_int_result::sortlength(THD *thd,
                                         const Type_std_attributes *item,
                                         SORT_FIELD_ATTR *sortorder) const
diff --git a/sql/item.cc b/sql/item.cc
index 99f801bb292..4b0aa2a348e 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1526,6 +1526,12 @@ String *Item_sp_variable::val_str(String *sp)
 }
 
 
+bool Item_sp_variable::val_native(THD *thd, Native *to)
+{
+  return val_native_from_item(thd, this_item(), to);
+}
+
+
 my_decimal *Item_sp_variable::val_decimal(my_decimal *decimal_value)
 {
   DBUG_ASSERT(fixed);
@@ -3178,6 +3184,18 @@ bool Item_field::get_date_result(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzyd
 }
 
 
+bool Item_field::val_native(THD *thd, Native *to)
+{
+  return val_native_from_field(field, to);
+}
+
+
+bool Item_field::val_native_result(THD *thd, Native *to)
+{
+  return val_native_from_field(result_field, to);
+}
+
+
 void Item_field::save_result(Field *to)
 {
   save_field_in_field(result_field, &null_value, to, TRUE);
@@ -4843,6 +4861,12 @@ String* Item_ref_null_helper::val_str(String* s)
 }
 
 
+bool Item_ref_null_helper::val_native(THD *thd, Native *to)
+{
+  return (owner->was_null|= val_native_from_item(thd, *ref, to));
+}
+
+
 bool Item_ref_null_helper::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 {  
   return (owner->was_null|= null_value= (*ref)->get_date_result(thd, ltime, fuzzydate));
@@ -8103,6 +8127,14 @@ String *Item_ref::str_result(String* str)
 }
 
 
+bool Item_ref::val_native_result(THD *thd, Native *to)
+{
+  return result_field ?
+         val_native_from_field(result_field, to) :
+         val_native(thd, to);
+}
+
+
 my_decimal *Item_ref::val_decimal_result(my_decimal *decimal_value)
 {
   if (result_field)
@@ -8197,6 +8229,12 @@ bool Item_ref::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 }
 
 
+bool Item_ref::val_native(THD *thd, Native *to)
+{
+  return val_native_from_item(thd, *ref, to);
+}
+
+
 my_decimal *Item_ref::val_decimal(my_decimal *decimal_value)
 {
   my_decimal *val= (*ref)->val_decimal_result(decimal_value);
@@ -8334,6 +8372,12 @@ bool Item_direct_ref::get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydat
 }
 
 
+bool Item_direct_ref::val_native(THD *thd, Native *to)
+{
+  return val_native_from_item(thd, *ref, to);
+}
+
+
 Item_cache_wrapper::~Item_cache_wrapper()
 {
   DBUG_ASSERT(expr_cache == 0);
@@ -8623,6 +8667,28 @@ String *Item_cache_wrapper::val_str(String* str)
 
 
 /**
+  Get the native value of the possibly cached item
+*/
+
+bool Item_cache_wrapper::val_native(THD *thd, Native* to)
+{
+  Item *cached_value;
+  DBUG_ENTER("Item_cache_wrapper::val_native");
+  if (!expr_cache)
+    DBUG_RETURN(val_native_from_item(thd, orig_item, to));
+
+  if ((cached_value= check_cache()))
+    DBUG_RETURN(val_native_from_item(thd, cached_value, to));
+
+  cache();
+  if ((null_value= expr_value->null_value))
+    DBUG_RETURN(true);
+  DBUG_RETURN(expr_value->val_native(thd, to));
+}
+
+
+
+/**
   Get the decimal value of the possibly cached item
 */
 
@@ -9794,6 +9860,117 @@ Item *Item_cache_time::make_literal(THD *thd)
   return new (thd->mem_root) Item_time_literal(thd, &ltime, decimals);
 }
 
+
+bool Item_cache_timestamp::val_native(THD *thd, Native *to)
+{
+  if (!has_value())
+  {
+    null_value= true;
+    return true;
+  }
+  return null_value= to->copy(m_native);
+}
+
+
+String *Item_cache_timestamp::val_str(String *str)
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return NULL;
+  }
+  return Datetime(this).to_string(str, decimals);
+}
+
+
+my_decimal *Item_cache_timestamp::val_decimal(my_decimal *decimal_value)
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return NULL;
+  }
+  return Datetime(this).to_decimal(decimal_value);
+}
+
+
+longlong Item_cache_timestamp::val_int()
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return 0;
+  }
+  return Datetime(this).to_longlong();
+}
+
+
+longlong Item_cache_timestamp::val_datetime_packed(THD *thd)
+{
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+longlong Item_cache_timestamp::val_time_packed(THD *thd)
+{
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+double Item_cache_timestamp::val_real()
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return 0;
+  }
+  return Datetime(this).to_double();
+}
+
+
+bool Item_cache_timestamp::get_date(THD *thd, MYSQL_TIME *ltime,
+                                    date_mode_t fuzzydate)
+{
+  if (!has_value())
+  {
+    set_zero_time(ltime, MYSQL_TIMESTAMP_DATETIME);
+    return true;
+  }
+  Timestamp_or_zero_datetime tm(m_native);
+  return (null_value= tm.to_TIME(thd, ltime, fuzzydate));
+}
+
+
+bool Item_cache_timestamp::cache_value()
+{
+  if (!example)
+    return false;
+  value_cached= true;
+  null_value= example->val_native_with_conversion_result(current_thd, &m_native,
+                                                         type_handler());
+  return true;
+}
+
+
+int Item_cache_timestamp::save_in_field(Field *field, bool no_conversions)
+{
+  if (!has_value())
+    return set_field_to_null_with_conversions(field, no_conversions);
+  if (!null_value && field->type_handler()->is_timestamp_type())
+  {
+    field->set_notnull();
+    return field->store_native(m_native);
+  }
+  return save_date_in_field(field, no_conversions);
+}
+
+
 bool Item_cache_real::cache_value()
 {
   if (!example)
diff --git a/sql/item.h b/sql/item.h
index 2ee2eabddac..4ddab28d5c3 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -855,6 +855,25 @@ protected:
       res= NULL;
     return res;
   }
+  bool val_native_from_item(THD *thd, Item *item, Native *to)
+  {
+    DBUG_ASSERT(is_fixed());
+    null_value= item->val_native(thd, to);
+    DBUG_ASSERT(null_value == item->null_value);
+    return null_value;
+  }
+  bool val_native_from_field(Field *field, Native *to)
+  {
+    if ((null_value= field->is_null()))
+      return true;
+    return (null_value= field->val_native(to));
+  }
+  bool val_native_with_conversion_from_item(THD *thd, Item *item, Native *to,
+                                            const Type_handler *handler)
+  {
+    DBUG_ASSERT(is_fixed());
+    return null_value= item->val_native_with_conversion(thd, to, handler);
+  }
   my_decimal *val_decimal_from_item(Item *item, my_decimal *decimal_value)
   {
     DBUG_ASSERT(is_fixed());
@@ -1276,6 +1295,60 @@ public:
   */
   virtual String *val_str(String *str)=0;
 
+
+  bool val_native_with_conversion(THD *thd, Native *to, const Type_handler *th)
+  {
+    return th->Item_val_native_with_conversion(thd, this, to);
+  }
+  bool val_native_with_conversion_result(THD *thd, Native *to,
+                                         const Type_handler *th)
+  {
+    return th->Item_val_native_with_conversion_result(thd, this, to);
+  }
+
+  virtual bool val_native(THD *thd, Native *to)
+  {
+   /*
+     The default implementation for the Items that do not need native format:
+     - Item_basic_value
+     - Item_ident_for_show
+     - Item_copy
+     - Item_exists_subselect
+     - Item_sum_field
+     - Item_sum_or_func (default implementation)
+     - Item_proc
+     - Item_type_holder (as val_xxx() are never called for it);
+     - TODO: Item_name_const will need val_native() in the future,
+       when we add this syntax:
+         TIMESTAMP WITH LOCAL TIMEZONE'2001-01-01 00:00:00'
+
+     These hybrid Item types override val_native():
+     - Item_field
+     - Item_param
+     - Item_sp_variable
+     - Item_ref
+     - Item_cache_wrapper
+     - Item_direct_ref
+     - Item_direct_view_ref
+     - Item_ref_null_helper
+     - Item_sum_or_func
+         Note, these hybrid type Item_sum_or_func descendants
+         override the default implementation:
+         * Item_sum_hybrid
+         * Item_func_hybrid_field_type
+         * Item_func_min_max
+         * Item_func_sp
+         * Item_func_last_value
+         * Item_func_rollup_const
+   */
+    DBUG_ASSERT(0);
+    return null_value= true;
+  }
+  virtual bool val_native_result(THD *thd, Native *to)
+  {
+    return val_native(thd, to);
+  }
+
   /*
     Returns string representation of this item in ASCII format.
 
@@ -2689,6 +2762,7 @@ public:
   String *val_str(String *sp);
   my_decimal *val_decimal(my_decimal *decimal_value);
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  bool val_native(THD *thd, Native *to);
   bool is_null();
 
 public:
@@ -3229,6 +3303,8 @@ public:
   void save_result(Field *to);
   double val_result();
   longlong val_int_result();
+  bool val_native(THD *thd, Native *to);
+  bool val_native_result(THD *thd, Native *to);
   String *str_result(String* tmp);
   my_decimal *val_decimal_result(my_decimal *);
   bool val_bool_result();
@@ -3815,6 +3891,11 @@ public:
     return can_return_value() ? value.val_str(str, this) : NULL;
   }
   bool get_date(THD *thd, MYSQL_TIME *tm, date_mode_t fuzzydate);
+  bool val_native(THD *thd, Native *to)
+  {
+    return Item_param::type_handler()->Item_param_val_native(thd, this, to);
+  }
+
   int  save_in_field(Field *field, bool no_conversions);
 
   void set_default();
@@ -4614,6 +4695,39 @@ public:
 };
 
 
+class Item_timestamp_literal: public Item_literal
+{
+  Timestamp_or_zero_datetime m_value;
+public:
+  Item_timestamp_literal(THD *thd)
+   :Item_literal(thd)
+  { }
+  const Type_handler *type_handler() const { return &type_handler_timestamp2; }
+  longlong val_int() { return Datetime(this).to_longlong(); }
+  double val_real() { return Datetime(this).to_double(); }
+  String *val_str(String *to)
+  {
+    return Datetime(this).to_string(to, decimals);
+  }
+  my_decimal *val_decimal(my_decimal *to)
+  {
+    return Datetime(this).to_decimal(to);
+  }
+  bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
+  {
+    bool res= m_value.to_TIME(thd, ltime, fuzzydate);
+    DBUG_ASSERT(!res);
+    return res;
+  }
+  void set_value(const Timestamp_or_zero_datetime *value)
+  {
+    m_value= *value;
+  }
+  Item *get_copy(THD *thd)
+  { return get_item_copy<Item_timestamp_literal>(thd, this); }
+};
+
+
 class Item_temporal_literal :public Item_literal
 {
 protected:
@@ -5062,11 +5176,13 @@ public:
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   String *val_str(String* tmp);
+  bool val_native(THD *thd, Native *to);
   bool is_null();
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
   double val_result();
   longlong val_int_result();
   String *str_result(String* tmp);
+  bool val_native_result(THD *thd, Native *to);
   my_decimal *val_decimal_result(my_decimal *);
   bool val_bool_result();
   bool is_null_result();
@@ -5271,6 +5387,7 @@ public:
   double val_real();
   longlong val_int();
   String *val_str(String* tmp);
+  bool val_native(THD *thd, Native *to);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool is_null();
@@ -5367,6 +5484,7 @@ public:
   double val_real();
   longlong val_int();
   String *val_str(String* tmp);
+  bool val_native(THD *thd, Native *to);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool is_null();
@@ -5563,6 +5681,12 @@ public:
     else
       return Item_direct_ref::val_str(tmp);
   }
+  bool val_native(THD *thd, Native *to)
+  {
+    if (check_null_ref())
+      return true;
+    return Item_direct_ref::val_native(thd, to);
+  }
   my_decimal *val_decimal(my_decimal *tmp)
   {
     if (check_null_ref())
@@ -5708,6 +5832,7 @@ public:
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  bool val_native(THD *thd, Native *to);
   virtual void print(String *str, enum_query_type query_type);
   table_map used_tables() const;
   Item *get_copy(THD *thd)
@@ -6531,6 +6656,27 @@ public:
 };
 
 
+class Item_cache_timestamp: public Item_cache
+{
+  Timestamp::NativeBuffer m_native;
+public:
+  Item_cache_timestamp(THD *thd)
+   :Item_cache(thd, &type_handler_timestamp2) { }
+  Item *get_copy(THD *thd)
+  { return get_item_copy<Item_cache_timestamp>(thd, this); }
+  bool cache_value();
+  String* val_str(String *str);
+  my_decimal *val_decimal(my_decimal *);
+  longlong val_int();
+  longlong val_datetime_packed(THD *thd);
+  longlong val_time_packed(THD *thd);
+  double val_real();
+  bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  int save_in_field(Field *field, bool no_conversions);
+  bool val_native(THD *thd, Native *to);
+};
+
+
 class Item_cache_real: public Item_cache
 {
   double value;
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 624d0f062cc..1f7d61bbe6a 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -567,6 +567,18 @@ bool Arg_comparator::set_cmp_func_datetime()
 }
 
 
+bool Arg_comparator::set_cmp_func_native()
+{
+  THD *thd= current_thd;
+  m_compare_collation= &my_charset_numeric;
+  func= is_owner_equal_func() ? &Arg_comparator::compare_e_native :
+                                &Arg_comparator::compare_native;
+  a= cache_converted_constant(thd, a, &a_cache, compare_type_handler());
+  b= cache_converted_constant(thd, b, &b_cache, compare_type_handler());
+  return false;
+}
+
+
 bool Arg_comparator::set_cmp_func_int()
 {
   THD *thd= current_thd;
@@ -770,6 +782,39 @@ int Arg_comparator::compare_e_string()
 }
 
 
+int Arg_comparator::compare_native()
+{
+  THD *thd= current_thd;
+  if (!(*a)->val_native_with_conversion(thd, &m_native1,
+                                        compare_type_handler()))
+  {
+    if (!(*b)->val_native_with_conversion(thd, &m_native2,
+                                          compare_type_handler()))
+    {
+      if (set_null)
+        owner->null_value= 0;
+      return compare_type_handler()->cmp_native(m_native1, m_native2);
+    }
+  }
+  if (set_null)
+    owner->null_value= 1;
+  return -1;
+}
+
+
+int Arg_comparator::compare_e_native()
+{
+  THD *thd= current_thd;
+  bool res1= (*a)->val_native_with_conversion(thd, &m_native1,
+                                              compare_type_handler());
+  bool res2= (*b)->val_native_with_conversion(thd, &m_native2,
+                                              compare_type_handler());
+  if (res1 || res2)
+    return MY_TEST(res1 == res2);
+  return MY_TEST(compare_type_handler()->cmp_native(m_native1, m_native2) == 0);
+}
+
+
 int Arg_comparator::compare_real()
 {
   /*
@@ -2121,6 +2166,29 @@ longlong Item_func_between::val_int_cmp_time()
 }
 
 
+longlong Item_func_between::val_int_cmp_native()
+{
+  THD *thd= current_thd;
+  const Type_handler *h= m_comparator.type_handler();
+  NativeBuffer<STRING_BUFFER_USUAL_SIZE> value, a, b;
+  if (val_native_with_conversion_from_item(thd, args[0], &value, h))
+    return 0;
+  bool ra= args[1]->val_native_with_conversion(thd, &a, h);
+  bool rb= args[2]->val_native_with_conversion(thd, &b, h);
+  if (!ra && !rb)
+    return (longlong)
+      ((h->cmp_native(value, a) >= 0 &&
+        h->cmp_native(value, b) <= 0) != negated);
+  if (ra && rb)
+    null_value= true;
+  else if (ra)
+    null_value= h->cmp_native(value, b) <= 0;
+  else
+    null_value= h->cmp_native(value, a) >= 0;
+  return (longlong) (!null_value && negated);
+}
+
+
 longlong Item_func_between::val_int_cmp_string()
 {
   String *value,*a,*b;
@@ -2306,6 +2374,15 @@ Item_func_ifnull::str_op(String *str)
 }
 
 
+bool Item_func_ifnull::native_op(THD *thd, Native *to)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (!val_native_with_conversion_from_item(thd, args[0], to, type_handler()))
+    return false;
+  return val_native_with_conversion_from_item(thd, args[1], to, type_handler());
+}
+
+
 bool Item_func_ifnull::date_op(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
 {
   DBUG_ASSERT(fixed == 1);
@@ -2829,6 +2906,16 @@ Item_func_nullif::time_op(THD *thd, MYSQL_TIME *ltime)
 
 
 bool
+Item_func_nullif::native_op(THD *thd, Native *to)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (!compare())
+    return (null_value= true);
+  return val_native_with_conversion_from_item(thd, args[2], to, type_handler());
+}
+
+
+bool
 Item_func_nullif::is_null()
 {
   return (null_value= (!compare() ? 1 : args[2]->is_null()));
@@ -3002,6 +3089,16 @@ bool Item_func_case::time_op(THD *thd, MYSQL_TIME *ltime)
 }
 
 
+bool Item_func_case::native_op(THD *thd, Native *to)
+{
+  DBUG_ASSERT(fixed == 1);
+  Item *item= find_item();
+  if (!item)
+    return (null_value= true);
+  return val_native_with_conversion_from_item(thd, item, to, type_handler());
+}
+
+
 bool Item_func_case::fix_fields(THD *thd, Item **ref)
 {
   bool res= Item_func::fix_fields(thd, ref);
@@ -3360,6 +3457,18 @@ bool Item_func_coalesce::time_op(THD *thd, MYSQL_TIME *ltime)
 }
 
 
+bool Item_func_coalesce::native_op(THD *thd, Native *to)
+{
+  DBUG_ASSERT(fixed == 1);
+  for (uint i= 0; i < arg_count; i++)
+  {
+    if (!val_native_with_conversion_from_item(thd, args[i], to, type_handler()))
+      return false;
+  }
+  return (null_value= true);
+}
+
+
 my_decimal *Item_func_coalesce::decimal_op(my_decimal *decimal_value)
 {
   DBUG_ASSERT(fixed == 1);
@@ -3637,6 +3746,53 @@ Item *in_longlong::create_item(THD *thd)
 }
 
 
+static int cmp_timestamp(void *cmp_arg,
+                         Timestamp_or_zero_datetime *a,
+                         Timestamp_or_zero_datetime *b)
+{
+  return a->cmp(*b);
+}
+
+
+in_timestamp::in_timestamp(THD *thd, uint elements)
+  :in_vector(thd, elements, sizeof(Value), (qsort2_cmp) cmp_timestamp, 0)
+{}
+
+
+void in_timestamp::set(uint pos, Item *item)
+{
+  Timestamp_or_zero_datetime *buff= &((Timestamp_or_zero_datetime *) base)[pos];
+  Timestamp::NativeBuffer native;
+  if (item->val_native_with_conversion(current_thd, &native, type_handler()))
+    *buff= Timestamp_or_zero_datetime();
+  else
+    *buff= Timestamp_or_zero_datetime(native);
+}
+
+
+uchar *in_timestamp::get_value(Item *item)
+{
+  Timestamp::NativeBuffer native;
+  if (item->val_native_with_conversion(current_thd, &native, type_handler()))
+    return 0;
+  tmp= Timestamp_or_zero_datetime(native);
+  return (uchar*) &tmp;
+}
+
+
+Item *in_timestamp::create_item(THD *thd)
+{
+  return new (thd->mem_root) Item_timestamp_literal(thd);
+}
+
+
+void in_timestamp::value_to_item(uint pos, Item *item)
+{
+  const Timestamp_or_zero_datetime *buff= &(((Timestamp_or_zero_datetime*) base)[pos]);
+  static_cast<Item_timestamp_literal*>(item)->set_value(buff);
+}
+
+
 void in_datetime::set(uint pos,Item *item)
 {
   struct packed_longlong *buff= &((packed_longlong*) base)[pos];
@@ -4044,6 +4200,50 @@ cmp_item *cmp_item_time::make_same()
 }
 
 
+void cmp_item_timestamp::store_value(Item *item)
+{
+  item->val_native_with_conversion(current_thd, &m_native,
+                                   &type_handler_timestamp2);
+  m_null_value= item->null_value;
+}
+
+
+int cmp_item_timestamp::cmp_not_null(const Value *val)
+{
+  /*
+    This method will be implemented when we add this syntax:
+      SELECT TIMESTAMP WITH LOCAL TIME ZONE '2001-01-01 10:20:30'
+    For now TIMESTAMP is compared to non-TIMESTAMP using DATETIME.
+  */
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+int cmp_item_timestamp::cmp(Item *arg)
+{
+  Timestamp::NativeBuffer tmp;
+  THD *thd= current_thd;
+  arg->val_native_with_conversion(thd, &tmp, &type_handler_timestamp2);
+  return m_null_value || arg->null_value ? UNKNOWN :
+         type_handler_timestamp2.cmp_native(m_native, tmp) != 0;
+}
+
+
+int cmp_item_timestamp::compare(cmp_item *arg)
+{
+  cmp_item_timestamp *tmp= static_cast<cmp_item_timestamp*>(arg);
+  return type_handler_timestamp2.cmp_native(m_native, tmp->m_native);
+}
+
+
+cmp_item* cmp_item_timestamp::make_same()
+{
+  return new cmp_item_timestamp();
+}
+
+
+
 bool Item_func_in::count_sargable_conds(void *arg)
 {
   ((SELECT_LEX*) arg)->cond_count++;
diff --git a/sql/item_cmpfunc.h b/sql/item_cmpfunc.h
index 4406eca7357..575922299df 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -68,6 +68,7 @@ class Arg_comparator: public Sql_alloc
     if (val1 == val2) return 0;
     return 1;
   }
+  NativeBuffer<STRING_BUFFER_USUAL_SIZE> m_native1, m_native2;
 public:
   /* Allow owner function to use string buffers. */
   String value1, value2;
@@ -89,6 +90,7 @@ public:
   bool set_cmp_func_string();
   bool set_cmp_func_time();
   bool set_cmp_func_datetime();
+  bool set_cmp_func_native();
   bool set_cmp_func_int();
   bool set_cmp_func_real();
   bool set_cmp_func_decimal();
@@ -121,6 +123,8 @@ public:
   int compare_e_datetime();
   int compare_time();
   int compare_e_time();
+  int compare_native();
+  int compare_e_native();
   int compare_json_str_basic(Item *j, Item *s);
   int compare_json_str();
   int compare_str_json();
@@ -935,6 +939,7 @@ public:
   longlong val_int_cmp_string();
   longlong val_int_cmp_datetime();
   longlong val_int_cmp_time();
+  longlong val_int_cmp_native();
   longlong val_int_cmp_int();
   longlong val_int_cmp_real();
   longlong val_int_cmp_decimal();
@@ -1013,6 +1018,7 @@ public:
   my_decimal *decimal_op(my_decimal *);
   bool date_op(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
   bool time_op(THD *thd, MYSQL_TIME *ltime);
+  bool native_op(THD *thd, Native *to);
   bool fix_length_and_dec()
   {
     if (aggregate_for_result(func_name(), args, arg_count, true))
@@ -1092,6 +1098,7 @@ public:
   my_decimal *decimal_op(my_decimal *);
   bool date_op(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
   bool time_op(THD *thd, MYSQL_TIME *ltime);
+  bool native_op(THD *thd, Native *to);
   bool fix_length_and_dec()
   {
     if (Item_func_case_abbreviation2::fix_length_and_dec2(args))
@@ -1150,6 +1157,11 @@ public:
   {
     return val_str_from_item(find_item(), str);
   }
+  bool native_op(THD *thd, Native *to)
+  {
+    return val_native_with_conversion_from_item(thd, find_item(), to,
+                                                type_handler());
+  }
 };
 
 
@@ -1249,6 +1261,7 @@ public:
   longlong int_op();
   String *str_op(String *str);
   my_decimal *decimal_op(my_decimal *);
+  bool native_op(THD *thd, Native *to);
   bool fix_length_and_dec();
   bool walk(Item_processor processor, bool walk_subquery, void *arg);
   const char *func_name() const { return "nullif"; }
@@ -1412,6 +1425,19 @@ public:
 };
 
 
+class in_timestamp :public in_vector
+{
+  Timestamp_or_zero_datetime tmp;
+public:
+  in_timestamp(THD *thd, uint elements);
+  void set(uint pos,Item *item);
+  uchar *get_value(Item *item);
+  Item* create_item(THD *thd);
+  void value_to_item(uint pos, Item *item);
+  const Type_handler *type_handler() const { return &type_handler_timestamp2; }
+};
+
+
 /*
   Class to represent a vector of constant DATE/DATETIME values.
 */
@@ -1666,6 +1692,20 @@ public:
   cmp_item *make_same();
 };
 
+
+class cmp_item_timestamp: public cmp_item_scalar
+{
+  Timestamp::NativeBuffer m_native;
+public:
+  cmp_item_timestamp() :cmp_item_scalar() { }
+  void store_value(Item *item);
+  int cmp_not_null(const Value *val);
+  int cmp(Item *arg);
+  int compare(cmp_item *ci);
+  cmp_item *make_same();
+};
+
+
 class cmp_item_real : public cmp_item_scalar
 {
   double value;
@@ -2132,6 +2172,7 @@ public:
   my_decimal *decimal_op(my_decimal *);
   bool date_op(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
   bool time_op(THD *thd, MYSQL_TIME *ltime);
+  bool native_op(THD *thd, Native *to);
   bool fix_fields(THD *thd, Item **ref);
   table_map not_null_tables() const { return 0; }
   const char *func_name() const { return "case"; }
diff --git a/sql/item_func.cc b/sql/item_func.cc
index ee2367a87b0..7c235a450e3 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -2730,6 +2730,36 @@ my_decimal *Item_func_min_max::val_decimal_native(my_decimal *dec)
 }
 
 
+bool Item_func_min_max::val_native(THD *thd, Native *native)
+{
+  DBUG_ASSERT(fixed == 1);
+  const Type_handler *handler= Item_hybrid_func::type_handler();
+  /*
+    For now we don't use val_native() with traditional data types other
+    than TIMESTAMP. If we ever start to use, we'll possibly need new methods:
+      Type_handler::Item_func_min_max_val_native()
+      Item::val_native_from_{int|decimal|real|str|date}()
+    It may happen to be cheaper to do the operation in e.g. INT format and then
+    convert INT to native (instead of doing the operation in native format).
+  */
+  NativeBuffer<STRING_BUFFER_USUAL_SIZE> cur;
+  for (uint i= 0; i < arg_count; i++)
+  {
+    if (val_native_with_conversion_from_item(thd, args[i],
+                                             i == 0 ? native : &cur,
+                                             handler))
+      return true;
+    if (i > 0)
+    {
+      int cmp= handler->cmp_native(*native, cur);
+      if ((cmp_sign < 0 ? cmp : -cmp) < 0 && native->copy(cur))
+        return null_value= true;
+    }
+  }
+  return null_value= false;
+}
+
+
 longlong Item_func_bit_length::val_int()
 {
   DBUG_ASSERT(fixed == 1);
@@ -6453,6 +6483,14 @@ String *Item_func_last_value::val_str(String *str)
   return tmp;
 }
 
+
+bool Item_func_last_value::val_native(THD *thd, Native *to)
+{
+  evaluate_sideeffects();
+  return val_native_from_item(thd, last_value, to);
+}
+
+
 longlong Item_func_last_value::val_int()
 {
   longlong tmp;
diff --git a/sql/item_func.h b/sql/item_func.h
index 660d39f48ea..1fc9f30b168 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -718,6 +718,15 @@ public:
   double val_real_from_time_op();
   double val_real_from_int_op();
 
+  /*
+    For now only the TIMESTAMP data type uses val_native().
+    For conversion purposes (e.g. to string and numbers),
+    it uses DATETIME representation.
+    Later (e.g. for INET6) we'll implement the following methods:
+      val_{str|decimal|int|real}_from_native_op()
+      get_date_from_native_op()
+  */
+
 public:
   Item_func_hybrid_field_type(THD *thd):
     Item_hybrid_func(thd)
@@ -768,6 +777,24 @@ public:
            Item_func_hybrid_field_type_get_date_with_warn(thd, this, to, mode);
   }
 
+  bool val_native(THD *thd, Native *to)
+  {
+    DBUG_ASSERT(fixed);
+    /*
+      For now only the TIMESTAMP data type uses val_native().
+      Later we'll probably need new methods:
+        Item_func_hybrid_field_type::val_native_from_{str|int|real|date}()
+        Type_handler::Item_func_hybrid_field_type_val_native(),
+      for symmetry with val_xxx() and get_date() above,
+      so the type handler can decide how to perform the hybrid operation
+      and when and how do data type conversion to native format.
+      It may happen to be cheaper to do the operation in e.g. INT format then
+      convert INT to native (instead of doing the operation in native format).
+      For now it's OK just to call native_op() directly here.
+    */
+    return native_op(thd, to);
+  }
+
   /**
      @brief Performs the operation that this functions implements when the
      result type is INT.
@@ -838,6 +865,7 @@ public:
   */
   virtual bool time_op(THD *thd, MYSQL_TIME *res)= 0;
 
+  virtual bool native_op(THD *thd, Native *native)= 0;
 };
 
 
@@ -905,6 +933,11 @@ public:
     DBUG_ASSERT(0);
     return true;
   }
+  bool native_op(THD *thd, Native *to)
+  {
+    DBUG_ASSERT(0);
+    return true;
+  }
 };
 
 
@@ -1771,6 +1804,7 @@ public:
     return Item_func_min_max::type_handler()->
              Item_func_min_max_get_date(thd, this, res, fuzzydate);
   }
+  bool val_native(THD *thd, Native *to);
   void aggregate_attributes_real(Item **items, uint nitems)
   {
     /*
@@ -1834,6 +1868,8 @@ public:
   double val_real() { return val_real_from_item(args[0]); }
   longlong val_int() { return val_int_from_item(args[0]); }
   String *val_str(String *str) { return val_str_from_item(args[0], str); }
+  bool val_native(THD *thd, Native *to)
+    { return val_native_from_item(thd, args[0], to); }
   my_decimal *val_decimal(my_decimal *dec)
     { return val_decimal_from_item(args[0], dec); }
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate)
@@ -3153,6 +3189,13 @@ public:
     return str;
   }
 
+  bool val_native(THD *thd, Native *to)
+  {
+    if (execute())
+      return true;
+    return null_value= sp_result_field->val_native(to);
+  }
+
   void update_null_value()
   { 
     execute();
@@ -3282,6 +3325,7 @@ public:
   String *val_str(String *);
   my_decimal *val_decimal(my_decimal *);
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
+  bool val_native(THD *thd, Native *);
   bool fix_length_and_dec();
   const char *func_name() const { return "last_value"; }
   const Type_handler *type_handler() const { return last_value->type_handler(); }
diff --git a/sql/item_subselect.cc b/sql/item_subselect.cc
index 4a1ed59d7df..9c0fd057d0f 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1352,6 +1352,24 @@ String *Item_singlerow_subselect::val_str(String *str)
 }
 
 
+bool Item_singlerow_subselect::val_native(THD *thd, Native *to)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (forced_const)
+    return value->val_native(thd, to);
+  if (!exec() && !value->null_value)
+  {
+    null_value= false;
+    return value->val_native(thd, to);
+  }
+  else
+  {
+    reset();
+    return true;
+  }
+}
+
+
 my_decimal *Item_singlerow_subselect::val_decimal(my_decimal *decimal_value)
 {
   DBUG_ASSERT(fixed == 1);
diff --git a/sql/item_subselect.h b/sql/item_subselect.h
index 443354f4900..0e771bae42e 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -306,6 +306,7 @@ public:
   double val_real();
   longlong val_int ();
   String *val_str (String *);
+  bool val_native(THD *thd, Native *);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index b571289ca12..fe901b65aeb 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -2381,6 +2381,15 @@ Item_sum_hybrid::val_str(String *str)
 }
 
 
+bool Item_sum_hybrid::val_native(THD *thd, Native *to)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (null_value)
+    return true;
+  return val_native_from_item(thd, value, to);
+}
+
+
 void Item_sum_hybrid::cleanup()
 {
   DBUG_ENTER("Item_sum_hybrid::cleanup");
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 01583d32f45..50e41990b13 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -1072,6 +1072,7 @@ protected:
   bool get_date(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate);
   void reset_field();
   String *val_str(String *);
+  bool val_native(THD *thd, Native *);
   const Type_handler *real_type_handler() const
   {
     return get_arg(0)->real_type_handler();
diff --git a/sql/item_timefunc.cc b/sql/item_timefunc.cc
index 1b728f561b7..8bdb3b41cdc 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -1218,14 +1218,16 @@ bool Item_func_unix_timestamp::get_timestamp_value(my_time_t *seconds,
   }
 
   THD *thd= current_thd;
-  Datetime dt(thd, args[0], Datetime::Options(TIME_NO_ZERO_IN_DATE, thd));
-  if ((null_value= !dt.is_valid_datetime()))
+  Timestamp::NativeBuffer native;
+  if (val_native_with_conversion_from_item(thd, args[0], &native,
+                                           &type_handler_timestamp2))
     return true;
-
-  uint error_code;
-  *seconds= TIME_to_timestamp(thd, dt.get_mysql_time(), &error_code);
-  *second_part= dt.get_mysql_time()->second_part;
-  return (null_value= (error_code == ER_WARN_DATA_OUT_OF_RANGE));
+  Timestamp_or_zero_datetime tm(native);
+  if (tm.is_zero_datetime())
+    return null_value= true;
+  *seconds= tm.tv().tv_sec;
+  *second_part= tm.tv().tv_usec;
+  return false;
 }
 
 
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index c99c1df20b6..b246d6e7e2e 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -209,6 +209,88 @@ Temporal_hybrid::Temporal_hybrid(THD *thd, Item *item, date_mode_t fuzzydate)
 }
 
 
+uint Timestamp::binary_length_to_precision(uint length)
+{
+  switch (length) {
+  case 4: return 0;
+  case 5: return 2;
+  case 6: return 4;
+  case 7: return 6;
+  }
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+Timestamp::Timestamp(const Native &native)
+{
+  DBUG_ASSERT(native.length() >= 4 && native.length() <= 7);
+  uint dec= binary_length_to_precision(native.length());
+  my_timestamp_from_binary(this, (const uchar *) native.ptr(), dec);
+}
+
+
+bool Timestamp::to_native(Native *to, uint decimals) const
+{
+  uint len= my_timestamp_binary_length(decimals);
+  if (to->reserve(len))
+    return true;
+  my_timestamp_to_binary(this, (uchar *) to->ptr(), decimals);
+  to->length(len);
+  return false;
+}
+
+
+bool Timestamp::to_TIME(THD *thd, MYSQL_TIME *to, date_mode_t fuzzydate) const
+{
+  return thd->timestamp_to_TIME(to, tv_sec, tv_usec, fuzzydate);
+}
+
+
+Timestamp::Timestamp(THD *thd, const MYSQL_TIME *ltime, uint *error_code)
+ :Timeval(TIME_to_timestamp(thd, ltime, error_code), ltime->second_part)
+{ }
+
+
+Timestamp_or_zero_datetime::Timestamp_or_zero_datetime(THD *thd,
+                                                       const MYSQL_TIME *ltime,
+                                                       uint *error_code)
+ :Timestamp(thd, ltime, error_code),
+  m_is_zero_datetime(*error_code == ER_WARN_DATA_OUT_OF_RANGE)
+{
+  if (m_is_zero_datetime)
+  {
+    if (!non_zero_date(ltime))
+      *error_code= 0;  // ltime was '0000-00-00 00:00:00'
+  }
+  else if (*error_code == ER_WARN_INVALID_TIMESTAMP)
+    *error_code= 0; // ltime fell into spring time gap, adjusted.
+}
+
+
+bool Timestamp_or_zero_datetime::to_TIME(THD *thd, MYSQL_TIME *to,
+                                         date_mode_t fuzzydate) const
+{
+  if (m_is_zero_datetime)
+  {
+    set_zero_time(to, MYSQL_TIMESTAMP_DATETIME);
+    return false;
+  }
+  return Timestamp::to_TIME(thd, to, fuzzydate);
+}
+
+
+bool Timestamp_or_zero_datetime::to_native(Native *to, uint decimals) const
+{
+  if (m_is_zero_datetime)
+  {
+    to->length(0);
+    return false;
+  }
+  return Timestamp::to_native(to, decimals);
+}
+
+
 void Sec6::make_from_decimal(const my_decimal *d, ulong *nanoseconds)
 {
   m_neg= my_decimal2seconds(d, &m_sec, &m_usec, nanoseconds);
@@ -1272,7 +1354,7 @@ const Type_handler *Type_handler_datetime_common::type_handler_for_comparison()
 
 const Type_handler *Type_handler_timestamp_common::type_handler_for_comparison() const
 {
-  return &type_handler_datetime;
+  return &type_handler_timestamp;
 }
 
 
@@ -1457,6 +1539,15 @@ Type_handler_hybrid_field_type::aggregate_for_comparison(const Type_handler *h)
       */
       if (b == TIME_RESULT)
         m_type_handler= h; // Temporal types bit non-temporal types
+      /*
+        Compare TIMESTAMP to a non-temporal type as DATETIME.
+        This is needed to make queries with fuzzy dates work:
+        SELECT * FROM t1
+        WHERE
+          ts BETWEEN '0000-00-00' AND '2010-00-01 00:00:00';
+      */
+      if (m_type_handler->is_timestamp_type())
+        m_type_handler= &type_handler_datetime;
     }
     else
     {
@@ -1540,7 +1631,16 @@ Type_handler_hybrid_field_type::aggregate_for_min_max(const Type_handler *h)
   }
   else if (a == TIME_RESULT || b == TIME_RESULT)
   {
-    if ((a == TIME_RESULT) + (b == TIME_RESULT) == 1)
+    if (m_type_handler->is_timestamp_type() + h->is_timestamp_type() == 1)
+    {
+      /*
+        Handle LEAST(TIMESTAMP, non-TIMESTAMP) as DATETIME,
+        to make sure fuzzy dates work in this context:
+          LEAST('2001-00-00', timestamp_field)
+      */
+      m_type_handler= &type_handler_datetime2;
+    }
+    else if ((a == TIME_RESULT) + (b == TIME_RESULT) == 1)
     {
       /*
         We're here if there's only one temporal data type:
@@ -3395,6 +3495,22 @@ int Type_handler_temporal_with_date::Item_save_in_field(Item *item,
 }
 
 
+int Type_handler_timestamp_common::Item_save_in_field(Item *item,
+                                                      Field *field,
+                                                      bool no_conversions)
+                                                      const
+{
+  Timestamp::NativeBuffer native;
+  if (Item_val_native_with_conversion(current_thd, item, &native))
+    return set_field_to_null_with_conversions(field, no_conversions);
+  field->set_notnull();
+  Timestamp_or_zero_datetime tm(native);
+  return tm.is_zero_datetime() ?
+         field->store_timestamp(0, 0) :
+         field->store_timestamp(tm.tv().tv_sec, tm.tv().tv_usec);
+}
+
+
 int Type_handler_string_result::Item_save_in_field(Item *item, Field *field,
                                                    bool no_conversions) const
 {
@@ -3461,6 +3577,12 @@ Type_handler_temporal_with_date::set_comparator_func(Arg_comparator *cmp) const
   return cmp->set_cmp_func_datetime();
 }
 
+bool
+Type_handler_timestamp_common::set_comparator_func(Arg_comparator *cmp) const
+{
+  return cmp->set_cmp_func_native();
+}
+
 
 /*************************************************************************/
 
@@ -3596,7 +3718,7 @@ Type_handler_string_result::Item_get_cache(THD *thd, const Item *item) const
 Item_cache *
 Type_handler_timestamp_common::Item_get_cache(THD *thd, const Item *item) const
 {
-  return new (thd->mem_root) Item_cache_datetime(thd);
+  return new (thd->mem_root) Item_cache_timestamp(thd);
 }
 
 Item_cache *
@@ -4763,6 +4885,12 @@ longlong Type_handler_time_common::
   return func->val_int_cmp_time();
 }
 
+longlong Type_handler_timestamp_common::
+           Item_func_between_val_int(Item_func_between *func) const
+{
+  return func->val_int_cmp_native();
+}
+
 longlong Type_handler_int_result::
            Item_func_between_val_int(Item_func_between *func) const
 {
@@ -4826,6 +4954,12 @@ cmp_item *Type_handler_temporal_with_date::make_cmp_item(THD *thd,
   return new (thd->mem_root) cmp_item_datetime;
 }
 
+cmp_item *Type_handler_timestamp_common::make_cmp_item(THD *thd,
+                                                       CHARSET_INFO *cs) const
+{
+  return new (thd->mem_root) cmp_item_timestamp;
+}
+
 /***************************************************************************/
 
 static int srtcmp_in(CHARSET_INFO *cs, const String *x,const String *y)
@@ -4886,6 +5020,15 @@ Type_handler_temporal_with_date::make_in_vector(THD *thd,
 }
 
 
+in_vector *
+Type_handler_timestamp_common::make_in_vector(THD *thd,
+                                              const Item_func_in *func,
+                                              uint nargs) const
+{
+  return new (thd->mem_root) in_timestamp(thd, nargs);
+}
+
+
 in_vector *Type_handler_row::make_in_vector(THD *thd,
                                             const Item_func_in *func,
                                             uint nargs) const
@@ -5004,7 +5147,11 @@ String *Type_handler_datetime_common::
 String *Type_handler_timestamp_common::
           Item_func_min_max_val_str(Item_func_min_max *func, String *str) const
 {
-  return Datetime(func).to_string(str, func->decimals);
+  MYSQL_TIME ltime;
+  if (get_date_from_val_native(current_thd, func, &ltime) ||
+      my_TIME_to_str(&ltime, str, func->decimals))
+    return NULL;
+  return str;
 }
 
 
@@ -5056,10 +5203,14 @@ double Type_handler_datetime_common::
   return Datetime(current_thd, func).to_double();
 }
 
+
 double Type_handler_timestamp_common::
          Item_func_min_max_val_real(Item_func_min_max *func) const
 {
-  return Datetime(current_thd, func).to_double();
+  MYSQL_TIME ltime;
+  if (get_date_from_val_native(current_thd, func, &ltime))
+    return 0;
+  return TIME_to_double(&ltime);
 }
 
 
@@ -5101,7 +5252,10 @@ longlong Type_handler_datetime_common::
 longlong Type_handler_timestamp_common::
          Item_func_min_max_val_int(Item_func_min_max *func) const
 {
-  return Datetime(current_thd, func).to_longlong();
+  MYSQL_TIME ltime;
+  if (get_date_from_val_native(current_thd, func, &ltime))
+    return 0;
+  return TIME_to_ulonglong(&ltime);
 }
 
 
@@ -5156,7 +5310,10 @@ my_decimal *Type_handler_timestamp_common::
             Item_func_min_max_val_decimal(Item_func_min_max *func,
                                           my_decimal *dec) const
 {
-  return Datetime(current_thd, func).to_decimal(dec);
+  MYSQL_TIME ltime;
+  if (get_date_from_val_native(current_thd, func, &ltime))
+    return 0;
+  return date2my_decimal(&ltime, dec);
 }
 
 
@@ -5208,6 +5365,14 @@ bool Type_handler_time_common::
   return func->get_time_native(thd, ltime);
 }
 
+
+bool Type_handler_timestamp_common::
+       Item_func_min_max_get_date(THD *thd, Item_func_min_max *func,
+                                  MYSQL_TIME *ltime, date_mode_t fuzzydate) const
+{
+  return get_date_from_val_native(thd, func, ltime);
+}
+
 /***************************************************************************/
 
 /**
@@ -6480,6 +6645,22 @@ bool Type_handler::
 }
 
 
+bool Type_handler::Item_send_timestamp(Item *item,
+                                       Protocol *protocol,
+                                       st_value *buf) const
+{
+  Timestamp::NativeBuffer native;
+  MYSQL_TIME ltime;
+  if (item->val_native(protocol->thd, &native))
+    return protocol->store_null();
+  Datetime::Options opt(TIME_CONV_NONE, protocol->thd);
+  return Timestamp_or_zero_datetime(native).
+           to_TIME(protocol->thd, &ltime, opt) ?
+         protocol->store_null() :
+         protocol->store(&ltime, item->decimals);
+}
+
+
 bool Type_handler::
        Item_send_datetime(Item *item, Protocol *protocol, st_value *buf) const
 {
@@ -7628,6 +7809,17 @@ bool Type_handler_temporal_with_date::Item_eq_value(THD *thd,
 }
 
 
+bool Type_handler_timestamp_common::Item_eq_value(THD *thd,
+                                                  const Type_cmp_attributes *attr,
+                                                  Item *a, Item *b) const
+{
+  Timestamp::NativeBuffer native0, native1;
+  a->val_native_with_conversion(thd, &native0, this);
+  b->val_native_with_conversion(thd, &native1, this);
+  return !a->null_value && !b->null_value && !cmp_native(native0, native1);
+}
+
+
 bool Type_handler_string_result::Item_eq_value(THD *thd,
                                                const Type_cmp_attributes *attr,
                                                Item *a, Item *b) const
@@ -7845,3 +8037,108 @@ Type_handler_time_common::create_literal_item(THD *thd,
   literal_warn(thd, item, str, length, cs, &st, "TIME", send_error);
   return item;
 }
+
+
+bool Type_handler_timestamp_common::TIME_to_native(THD *thd,
+                                                   const MYSQL_TIME *ltime,
+                                                   Native *to,
+                                                   uint decimals) const
+{
+  uint error_code;
+  Timestamp_or_zero_datetime tm(thd, ltime, &error_code);
+  if (error_code)
+    return true;
+  tm.trunc(decimals);
+  return tm.to_native(to, decimals);
+}
+
+
+bool
+Type_handler_timestamp_common::Item_val_native_with_conversion(THD *thd,
+                                                               Item *item,
+                                                               Native *to) const
+{
+  MYSQL_TIME ltime;
+  if (item->type_handler()->is_timestamp_type())
+    return item->val_native(thd, to);
+  return
+    item->get_date(thd, &ltime, Datetime::Options(TIME_NO_ZERO_IN_DATE, thd)) ||
+    TIME_to_native(thd, &ltime, to, item->datetime_precision(thd));
+}
+
+
+bool
+Type_handler_timestamp_common::Item_val_native_with_conversion_result(THD *thd,
+                                                                      Item *item,
+                                                                      Native *to)
+                                                                      const
+{
+  MYSQL_TIME ltime;
+  if (item->type_handler()->is_timestamp_type())
+    return item->val_native_result(thd, to);
+  return
+    item->get_date_result(thd, &ltime,
+                          Datetime::Options(TIME_NO_ZERO_IN_DATE, thd)) ||
+    TIME_to_native(thd, &ltime, to, item->datetime_precision(thd));
+}
+
+
+int Type_handler_timestamp_common::cmp_native(const Native &a,
+                                              const Native &b) const
+{
+  /*
+    Optimize a simple case:
+    Either both timeatamp values have the same fractional precision,
+    or both values are zero datetime '0000-00-00 00:00:00.000000',
+  */
+  if (a.length() == b.length())
+    return memcmp(a.ptr(), b.ptr(), a.length());
+  return Timestamp_or_zero_datetime(a).cmp(Timestamp_or_zero_datetime(b));
+}
+
+
+bool
+Type_handler_timestamp_common::get_date_from_val_native(THD *thd,
+                                                        Item *item,
+                                                        MYSQL_TIME *ltime)
+                                                        const
+{
+  Timestamp::NativeBuffer native;
+  if (item->val_native(thd, &native))
+  {
+    DBUG_ASSERT(item->null_value);
+    return true;
+  }
+  return item->null_value=
+    Timestamp_or_zero_datetime(native).
+     to_TIME(thd, ltime, Datetime::Options(TIME_CONV_NONE, thd));
+}
+
+
+bool
+Type_handler::Item_param_val_native(THD *thd,
+                                        Item_param *item,
+                                        Native *to) const
+{
+  DBUG_ASSERT(0); // TODO-TYPE: MDEV-14271
+  return item->null_value= true;
+}
+
+
+bool
+Type_handler_timestamp_common::Item_param_val_native(THD *thd,
+                                                         Item_param *item,
+                                                         Native *to) const
+{
+  /*
+    The below code may not run well in corner cases.
+    This will be fixed under terms of MDEV-14271.
+    Item_param should:
+    - either remember @@time_zone at bind time
+    - or store TIMESTAMP in my_time_t format, rather than in MYSQL_TIME format.
+  */
+  MYSQL_TIME ltime;
+  return
+    item->get_date(thd, &ltime, Datetime::Options(TIME_NO_ZERO_IN_DATE, thd)) ||
+    TIME_to_native(thd, &ltime, to, item->datetime_precision(thd));
+}
diff --git a/sql/sql_type.h b/sql/sql_type.h
index a5d8f4ae6df..3cbcfb7b62c 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -25,6 +25,7 @@
 #include "sql_array.h"
 #include "sql_const.h"
 #include "sql_time.h"
+#include "compat56.h"
 
 class Field;
 class Column_definition;
@@ -90,6 +91,25 @@ enum scalar_comparison_op
 };
 
 
+class Native: public Binary_string
+{
+public:
+  Native(char *str, size_t len)
+   :Binary_string(str, len)
+  { }
+};
+
+
+template<size_t buff_sz>
+class NativeBuffer: public Native
+{
+  char buff[buff_sz];
+public:
+  NativeBuffer() : Native(buff, buff_sz) { length(0); }
+};
+
+
+
 class Dec_ptr
 {
 protected:
@@ -2208,6 +2228,7 @@ public:
 
 class Timestamp: protected Timeval
 {
+  static uint binary_length_to_precision(uint length);
 protected:
   void round_or_set_max(uint dec, int *warn);
   bool add_nanoseconds_usec(uint nanoseconds)
@@ -2234,11 +2255,28 @@ public:
      :DatetimeOptions(sql_mode_for_timestamp(thd), default_round_mode(thd))
     { }
   };
+  class NativeBuffer: public ::NativeBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE>
+  {
+  };
 public:
   Timestamp(my_time_t timestamp, ulong sec_part)
    :Timeval(timestamp, sec_part)
   { }
+  explicit Timestamp(const struct timeval &tv)
+   :Timeval(tv)
+  { }
+  explicit Timestamp(const Native &native);
+  Timestamp(THD *thd, const MYSQL_TIME *ltime, uint *error_code);
   const struct timeval &tv() const { return *this; }
+  int cmp(const Timestamp &other) const
+  {
+    return tv_sec < other.tv_sec   ? -1 :
+           tv_sec > other.tv_sec   ? +1 :
+           tv_usec < other.tv_usec ? -1 :
+           tv_usec > other.tv_usec ? +1 : 0;
+  }
+  bool to_TIME(THD *thd, MYSQL_TIME *ltime, date_mode_t fuzzydate) const;
+  bool to_native(Native *to, uint decimals) const;
   long fraction_remainder(uint dec) const
   {
     return my_time_fraction_remainder(tv_usec, dec);
@@ -2274,6 +2312,59 @@ public:
 };
 
 
+/**
+  A helper class to store MariaDB TIMESTAMP values, which can be:
+  - real TIMESTAMP (seconds and microseconds since epoch), or
+  - zero datetime '0000-00-00 00:00:00.000000'
+*/
+class Timestamp_or_zero_datetime: public Timestamp
+{
+  bool m_is_zero_datetime;
+public:
+  Timestamp_or_zero_datetime()
+   :Timestamp(0,0), m_is_zero_datetime(true)
+  { }
+  Timestamp_or_zero_datetime(const Native &native)
+   :Timestamp(native.length() ? Timestamp(native) : Timestamp(0,0)),
+    m_is_zero_datetime(native.length() == 0)
+  { }
+  Timestamp_or_zero_datetime(const Timestamp &tm, bool is_zero_datetime)
+   :Timestamp(tm), m_is_zero_datetime(is_zero_datetime)
+  { }
+  Timestamp_or_zero_datetime(THD *thd, const MYSQL_TIME *ltime, uint *err_code);
+  bool is_zero_datetime() const { return m_is_zero_datetime; }
+  const struct timeval &tv() const
+  {
+    DBUG_ASSERT(!is_zero_datetime());
+    return Timestamp::tv();
+  }
+  void trunc(uint decimals)
+  {
+    if (!is_zero_datetime())
+     Timestamp::trunc(decimals);
+  }
+  int cmp(const Timestamp_or_zero_datetime &other) const
+  {
+    if (is_zero_datetime())
+      return other.is_zero_datetime() ? 0 : -1;
+    if (other.is_zero_datetime())
+      return 1;
+    return Timestamp::cmp(other);
+  }
+  bool to_TIME(THD *thd, MYSQL_TIME *to, date_mode_t fuzzydate) const;
+  /*
+    Convert to native format:
+    - Real timestamps are encoded in the same way how Field_timestamp2 stores
+      values (big endian seconds followed by big endian microseconds)
+    - Zero datetime '0000-00-00 00:00:00.000000' is encoded as empty string.
+    Two native values are binary comparable.
+  */
+  bool to_native(Native *to, uint decimals) const;
+};
+
+
+
+
 /*
   Flags for collation aggregation modes, used in TDCollation::agg():
 
@@ -2906,6 +2997,7 @@ protected:
   bool Item_send_double(Item *item, Protocol *protocol, st_value *buf) const;
   bool Item_send_time(Item *item, Protocol *protocol, st_value *buf) const;
   bool Item_send_date(Item *item, Protocol *protocol, st_value *buf) const;
+  bool Item_send_timestamp(Item *item, Protocol *protocol, st_value *buf) const;
   bool Item_send_datetime(Item *item, Protocol *protocol, st_value *buf) const;
   bool Column_definition_prepare_stage2_legacy(Column_definition *c,
                                                enum_field_types type)
@@ -3209,6 +3301,9 @@ public:
                                          Item_param *param,
                                          const Type_all_attributes *attr,
                                          const st_value *value) const= 0;
+  virtual bool Item_param_val_native(THD *thd,
+                                         Item_param *item,
+                                         Native *to) const;
   virtual bool Item_send(Item *item, Protocol *p, st_value *buf) const= 0;
   virtual int Item_save_in_field(Item *item, Field *field,
                                  bool no_conversions) const= 0;
@@ -3311,6 +3406,11 @@ public:
     DBUG_ASSERT(0);
     return NULL;
   }
+  virtual int cmp_native(const Native &a, const Native &b) const
+  {
+    DBUG_ASSERT(0);
+    return 0;
+  }
   virtual bool set_comparator_func(Arg_comparator *cmp) const= 0;
   virtual bool Item_const_eq(const Item_const *a, const Item_const *b,
                              bool binary_cmp) const
@@ -3335,6 +3435,17 @@ public:
   virtual
   bool Item_sum_variance_fix_length_and_dec(Item_sum_variance *) const= 0;
 
+  virtual bool Item_val_native_with_conversion(THD *thd, Item *item,
+                                               Native *to) const
+  {
+    return true;
+  }
+  virtual bool Item_val_native_with_conversion_result(THD *thd, Item *item,
+                                                      Native *to) const
+  {
+    return true;
+  }
+
   virtual bool Item_val_bool(Item *item) const= 0;
   virtual void Item_get_date(THD *thd, Item *item,
                              Temporal::Warn *buff, MYSQL_TIME *ltime,
@@ -5188,6 +5299,9 @@ public:
 class Type_handler_timestamp_common: public Type_handler_temporal_with_date
 {
   static const Name m_name_timestamp;
+protected:
+  bool TIME_to_native(THD *, const MYSQL_TIME *from, Native *to, uint dec) const;
+  bool get_date_from_val_native(THD *thd, Item *item, MYSQL_TIME *to) const;
 public:
   virtual ~Type_handler_timestamp_common() {}
   const Name name() const { return m_name_timestamp; }
@@ -5202,6 +5316,20 @@ public:
     return true;
   }
   void Column_definition_implicit_upgrade(Column_definition *c) const;
+  bool Item_eq_value(THD *thd, const Type_cmp_attributes *attr,
+                     Item *a, Item *b) const;
+  bool Item_val_native_with_conversion(THD *thd, Item *, Native *to) const;
+  bool Item_val_native_with_conversion_result(THD *thd, Item *, Native *to) const;
+  bool Item_param_val_native(THD *thd, Item_param *item, Native *to) const;
+  int cmp_native(const Native &a, const Native &b) const;
+  longlong Item_func_between_val_int(Item_func_between *func) const;
+  cmp_item *make_cmp_item(THD *thd, CHARSET_INFO *cs) const;
+  in_vector *make_in_vector(THD *thd, const Item_func_in *f, uint nargs) const;
+  void make_sort_key(uchar *to, Item *item, const SORT_FIELD_ATTR *sort_field,
+                     Sort_param *param) const;
+  void sortlength(THD *thd,
+                  const Type_std_attributes *item,
+                  SORT_FIELD_ATTR *attr) const;
   bool Column_definition_fix_attributes(Column_definition *c) const;
   uint Item_decimal_scale(const Item *item) const
   {
@@ -5214,8 +5342,9 @@ public:
   }
   bool Item_send(Item *item, Protocol *protocol, st_value *buf) const
   {
-    return Item_send_datetime(item, protocol, buf);
+    return Item_send_timestamp(item, protocol, buf);
   }
+  int Item_save_in_field(Item *item, Field *field, bool no_conversions) const;
   String *print_item_value(THD *thd, Item *item, String *str) const;
   Item_cache *Item_get_cache(THD *thd, const Item *item) const;
   String *Item_func_min_max_val_str(Item_func_min_max *, String *) const;
@@ -5223,6 +5352,7 @@ public:
   longlong Item_func_min_max_val_int(Item_func_min_max *) const;
   my_decimal *Item_func_min_max_val_decimal(Item_func_min_max *,
                                             my_decimal *) const;
+  bool set_comparator_func(Arg_comparator *cmp) const;
   bool Item_hybrid_func_fix_attributes(THD *thd,
                                        const char *name,
                                        Type_handler_hybrid_field_type *,
@@ -5230,6 +5360,8 @@ public:
                                        Item **items, uint nitems) const;
   void Item_param_set_param_func(Item_param *param,
                                  uchar **pos, ulong len) const;
+  bool Item_func_min_max_get_date(THD *thd, Item_func_min_max*,
+                                  MYSQL_TIME *, date_mode_t fuzzydate) const;
 };
 
 
diff --git a/sql/structs.h b/sql/structs.h
index dec1e4de0c7..8728dee1918 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -847,12 +847,17 @@ public:
 
 class Timeval: public timeval
 {
+protected:
+  Timeval() { }
 public:
   Timeval(my_time_t sec, ulong usec)
   {
     tv_sec= sec;
     tv_usec= usec;
   }
+  explicit Timeval(const timeval &tv)
+   :timeval(tv)
+  { }
 };