diff --git a/mysql-test/main/timezone2.result b/mysql-test/main/timezone2.result
index 096e996..7b14990 100644
--- a/mysql-test/main/timezone2.result
+++ b/mysql-test/main/timezone2.result
@@ -332,3 +332,198 @@ NULL
 #
 # End of 5.3 tests
 #
+#
+# Start of 10.4 tests
+#
+#
+# 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 7a38610..4103959 100644
--- a/mysql-test/main/timezone2.test
+++ b/mysql-test/main/timezone2.test
@@ -308,3 +308,180 @@ SELECT CONVERT_TZ('2001-10-08 00:00:00', MAKE_SET(0,'+01:00'), '+00:00' );
 --echo #
 --echo # End of 5.3 tests
 --echo #
+
+
+--echo #
+--echo # Start of 10.4 tests
+--echo #
+
+
+--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 b0405bc..564f8d7 100644
--- a/mysql-test/main/type_timestamp.result
+++ b/mysql-test/main/type_timestamp.result
@@ -1014,3 +1014,29 @@ DROP TABLE t1;
 #
 # End of 10.3 tests
 #
+#
+# Start of 10.4 tests
+#
+#
+# MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+#
+# Testing Item_func_rollup_const::val_raw_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 6d81a86..6cd3b77 100644
--- a/mysql-test/main/type_timestamp.test
+++ b/mysql-test/main/type_timestamp.test
@@ -602,7 +602,35 @@ SHOW CREATE TABLE t2;
 DROP TABLE t2;
 DROP TABLE t1;
 
-
 --echo #
 --echo # End of 10.3 tests
 --echo #
+
+
+--echo #
+--echo # Start of 10.4 tests
+--echo #
+
+--echo #
+--echo # MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+--echo #
+
+--echo # Testing Item_func_rollup_const::val_raw_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 bb5e267..5cd150f 100644
--- a/sql/compat56.h
+++ b/sql/compat56.h
@@ -19,6 +19,15 @@
 
 
 /** MySQL56 routines and macros **/
+
+/*
+  Buffer size for a raw TIMESTAMP representation, for use with StringBuffer.
+  4 bytes for seconds
+  3 bytes for microseconds
+  1 byte for the trailing '\0' (as String always 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 dc85482..38ebe67 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -5060,6 +5060,22 @@ my_time_t Field_timestamp::get_timestamp(const uchar *pos,
 }
 
 
+bool Field_timestamp::val_raw_native(String *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, MYSQL_TIME *l_time,
                                              const ErrConv *str,
                                              int was_cut,
@@ -5448,6 +5464,17 @@ my_time_t Field_timestamp_hires::get_timestamp(const uchar *pos,
   return mi_uint4korr(pos);
 }
 
+
+bool Field_timestamp_hires::val_raw_native(String *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(tm.tv_sec == 0, tm).to_raw(to, dec);
+}
+
+
 double Field_timestamp_with_dec::val_real(void)
 {
   MYSQL_TIME ltime;
diff --git a/sql/field.h b/sql/field.h
index b6f2880..cc9fa9e 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -779,6 +779,15 @@ class Field: public Value_source
   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 raw format
+  */
+  virtual int store_raw_native(const char *str, uint length)
+  {
+    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,
@@ -820,6 +829,11 @@ class Field: public Value_source
      This trickery is used to decrease a number of malloc calls.
   */
   virtual String *val_str(String*,String *)=0;
+  virtual bool val_raw_native(String *to)
+  {
+    DBUG_ASSERT(!is_null());
+    return to->copy((const char *) ptr, pack_length(), &my_charset_bin);
+  }
   String *val_int_as_str(String *val_buffer, bool unsigned_flag);
   /*
     Return the field value as a LEX_CSTRING, without padding to full length
@@ -2607,6 +2621,20 @@ class Field_timestamp :public Field_temporal {
   {
     int4store(ptr,timestamp);
   }
+  void store_timeval(const struct timeval &tm)
+  {
+    store_TIME(tm.tv_sec, tm.tv_usec);
+  }
+  int store_raw_native(const char *str, uint length)
+  {
+    Timestamp_or_zero_datetime tm(str, length);
+    if (tm.is_zero_datetime())
+      reset();
+    else
+      store_timeval(tm.tm());
+    return 0;
+  }
+  bool val_raw_native(String *to);
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
   uchar *pack(uchar *to, const uchar *from,
               uint max_length __attribute__((unused)))
@@ -2686,6 +2714,7 @@ class Field_timestamp_hires :public Field_timestamp_with_dec {
   {
     DBUG_ASSERT(dec);
   }
+  bool val_raw_native(String *to);
   my_time_t get_timestamp(const uchar *pos, ulong *sec_part) const;
   void store_TIME(my_time_t timestamp, ulong sec_part);
   int cmp(const uchar *,const uchar *);
@@ -2737,6 +2766,16 @@ class Field_timestampf :public Field_timestamp_with_dec {
   {
     return get_timestamp(ptr, sec_part);
   }
+  bool val_raw_native(String *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_raw_native(to);
+  }
   uint size_of() const { return sizeof(*this); }
 };
 
diff --git a/sql/filesort.cc b/sql/filesort.cc
index e62c4a9..99cdf11 100644
--- a/sql/filesort.cc
+++ b/sql/filesort.cc
@@ -1069,6 +1069,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
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  uint binlen= my_timestamp_binary_length(item->decimals);
+  if (item->val_raw_native(&raw) || raw.length() == 0)
+  {
+    // NULL or '0000-00-00 00:00:00'
+    bzero(to, item->maybe_null ? binlen + 1 : binlen);
+  }
+  else
+  {
+    DBUG_ASSERT(raw.length() == binlen);
+    if (item->maybe_null)
+      *to++= 1;
+    memcpy((char *) to, raw.ptr(), binlen);
+  }
+}
+
+
+void
 Type_handler::make_sort_key_longlong(uchar *to,
                                      bool maybe_null,
                                      bool null_value,
@@ -1876,6 +1898,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 68aed25..d4b7586 100644
--- a/sql/item.cc
+++ b/sql/item.cc
@@ -1755,6 +1755,12 @@ String *Item_sp_variable::val_str(String *sp)
 }
 
 
+bool Item_sp_variable::val_raw_native(String *raw)
+{
+  return val_raw_native_from_item(this_item(), raw);
+}
+
+
 my_decimal *Item_sp_variable::val_decimal(my_decimal *decimal_value)
 {
   DBUG_ASSERT(fixed);
@@ -3432,6 +3438,18 @@ bool Item_field::get_date_result(MYSQL_TIME *ltime, ulonglong fuzzydate)
 }
 
 
+bool Item_field::val_raw_native(String *raw)
+{
+  return val_raw_native_from_field(field, raw);
+}
+
+
+bool Item_field::val_raw_native_result(String *raw)
+{
+  return val_raw_native_from_field(result_field, raw);
+}
+
+
 void Item_field::save_result(Field *to)
 {
   save_field_in_field(result_field, &null_value, to, TRUE);
@@ -5257,6 +5275,12 @@ String* Item_ref_null_helper::val_str(String* s)
 }
 
 
+bool Item_ref_null_helper::val_raw_native(String *raw)
+{
+  return (owner->was_null|= val_raw_native_from_item(*ref, raw));
+}
+
+
 bool Item_ref_null_helper::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
 {  
   return (owner->was_null|= null_value= (*ref)->get_date_result(ltime, fuzzydate));
@@ -8485,6 +8509,14 @@ String *Item_ref::str_result(String* str)
 }
 
 
+bool Item_ref::val_raw_native_result(String *raw)
+{
+  return result_field ?
+         val_raw_native_from_field(result_field, raw) :
+         val_raw_native(raw);
+}
+
+
 my_decimal *Item_ref::val_decimal_result(my_decimal *decimal_value)
 {
   if (result_field)
@@ -8579,6 +8611,12 @@ bool Item_ref::get_date(MYSQL_TIME *ltime,ulonglong fuzzydate)
 }
 
 
+bool Item_ref::val_raw_native(String *raw)
+{
+  return val_raw_native_from_item(*ref, raw);
+}
+
+
 my_decimal *Item_ref::val_decimal(my_decimal *decimal_value)
 {
   my_decimal *val= (*ref)->val_decimal_result(decimal_value);
@@ -8716,6 +8754,12 @@ bool Item_direct_ref::get_date(MYSQL_TIME *ltime,ulonglong fuzzydate)
 }
 
 
+bool Item_direct_ref::val_raw_native(String *raw)
+{
+  return val_raw_native_from_item(*ref, raw);
+}
+
+
 Item_cache_wrapper::~Item_cache_wrapper()
 {
   DBUG_ASSERT(expr_cache == 0);
@@ -9005,6 +9049,28 @@ String *Item_cache_wrapper::val_str(String* str)
 
 
 /**
+  Get the raw value of the possibly cached item
+*/
+
+bool Item_cache_wrapper::val_raw_native(String* raw)
+{
+  Item *cached_value;
+  DBUG_ENTER("Item_cache_wrapper::val_raw_native");
+  if (!expr_cache)
+    DBUG_RETURN(val_raw_native_from_item(orig_item, raw));
+
+  if ((cached_value= check_cache()))
+    DBUG_RETURN(val_raw_native_from_item(cached_value, raw));
+
+  cache();
+  if ((null_value= expr_value->null_value))
+    DBUG_RETURN(true);
+  DBUG_RETURN(expr_value->val_raw_native(raw));
+}
+
+
+
+/**
   Get the decimal value of the possibly cached item
 */
 
@@ -10283,6 +10349,115 @@ Item *Item_cache_time::make_literal(THD *thd)
   return new (thd->mem_root) Item_time_literal(thd, &ltime, decimals);
 }
 
+
+bool Item_cache_timestamp::val_raw_native(String *raw)
+{
+  if (!has_value())
+  {
+    null_value= true;
+    return true;
+  }
+  return null_value= raw->copy(m_raw.ptr(), m_raw.length(), &my_charset_bin);
+}
+
+
+String *Item_cache_timestamp::val_str(String *str)
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return NULL;
+  }
+  return val_string_from_date(str);
+}
+
+
+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 val_decimal_from_date(decimal_value);
+}
+
+
+longlong Item_cache_timestamp::val_int()
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return 0;
+  }
+  return val_int_from_date();
+}
+
+
+longlong Item_cache_timestamp::val_datetime_packed()
+{
+  DBUG_ASSERT(0);
+  return 0;
+}
+
+
+longlong Item_cache_timestamp::val_time_packed()
+{
+  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 val_real_from_date();
+}
+
+
+bool Item_cache_timestamp::get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
+{
+  if (!has_value())
+  {
+    set_zero_time(ltime, MYSQL_TIMESTAMP_DATETIME);
+    return true;
+  }
+  Timestamp_or_zero_datetime tm(&m_raw);
+  return (null_value= tm.to_TIME(current_thd, ltime, fuzzydate));
+}
+
+
+bool Item_cache_timestamp::cache_value()
+{
+  if (!example)
+    return false;
+  value_cached= true;
+  null_value= example->val_raw_with_conversion_result(&m_raw, 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_raw_native(m_raw.ptr(), m_raw.length());
+  }
+  return save_date_in_field(field, no_conversions);
+}
+
+
 bool Item_cache_real::cache_value()
 {
   if (!example)
@@ -10301,6 +10476,7 @@ double Item_cache_real::val_real()
   return value;
 }
 
+
 longlong Item_cache_real::val_int()
 {
   if (!has_value())
diff --git a/sql/item.h b/sql/item.h
index f6f4684..9531b9d 100644
--- a/sql/item.h
+++ b/sql/item.h
@@ -849,6 +849,25 @@ class Item: public Value_source,
       res= NULL;
     return res;
   }
+  bool val_raw_native_from_item(Item *item, String *raw)
+  {
+    DBUG_ASSERT(is_fixed());
+    null_value= item->val_raw_native(raw);
+    DBUG_ASSERT(null_value == item->null_value);
+    return null_value;
+  }
+  bool val_raw_native_from_field(Field *field, String *raw)
+  {
+    if ((null_value= field->is_null()))
+      return true;
+    return (null_value= field->val_raw_native(raw));
+  }
+  bool val_raw_with_conversion_from_item(Item *item, String *raw,
+                                         const Type_handler *handler)
+  {
+    DBUG_ASSERT(is_fixed());
+    return null_value= item->val_raw_with_conversion(raw, handler);
+  }
   my_decimal *val_decimal_from_item(Item *item, my_decimal *decimal_value)
   {
     DBUG_ASSERT(is_fixed());
@@ -1251,6 +1270,57 @@ class Item: public Value_source,
   */
   virtual String *val_str(String *str)=0;
 
+
+  bool val_raw_with_conversion(String *raw, const Type_handler *handler)
+  {
+    return handler->Item_val_raw_with_conversion(this, raw);
+  }
+  bool val_raw_with_conversion_result(String *raw, const Type_handler *handler)
+  {
+    return handler->Item_val_raw_with_conversion_result(this, raw);
+  }
+
+  virtual bool val_raw_native(String *str)
+  {
+   /*
+     The default implementation for the Items that do not need raw 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: TIMESTAMP WITH LOCAL TIMEZONE'2001-01-01 00:00:00
+
+     These hybrid Item types override val_raw_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_raw_native_result(String *str)
+  {
+    return val_raw_native(str);
+  }
+
   /*
     Returns string representation of this item in ASCII format.
 
@@ -2711,6 +2781,7 @@ class Item_sp_variable :public Item_fixed_hybrid
   String *val_str(String *sp);
   my_decimal *val_decimal(my_decimal *decimal_value);
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  bool val_raw_native(String *raw);
   bool is_null();
 
 public:
@@ -3251,6 +3322,8 @@ class Item_field :public Item_ident,
   void save_result(Field *to);
   double val_result();
   longlong val_int_result();
+  bool val_raw_native(String *str);
+  bool val_raw_native_result(String *str);
   String *str_result(String* tmp);
   my_decimal *val_decimal_result(my_decimal *);
   bool val_bool_result();
@@ -3836,6 +3909,11 @@ class Item_param :public Item_basic_value,
   {
     return can_return_value() ? value.val_str(str, this) : NULL;
   }
+  bool val_raw_native(String *raw)
+  {
+    return Item_param::type_handler()->Item_param_val_raw_native(this, raw);
+  }
+
   bool get_date(MYSQL_TIME *tm, ulonglong fuzzydate);
   int  save_in_field(Field *field, bool no_conversions);
 
@@ -4650,6 +4728,33 @@ class Item_bin_string: public Item_hex_hybrid
 };
 
 
+class Item_timestamp_literal: public Item_literal
+{
+  Timestamp_or_zero_datetime m_value;
+public:
+  Item_timestamp_literal(THD *thd)
+   :Item_literal(thd), m_value("",0)
+  { }
+  const Type_handler *type_handler() const { return &type_handler_timestamp2; }
+  double val_real() { return val_real_from_date(); }
+  longlong val_int() { return val_int_from_date(); }
+  my_decimal *val_decimal(my_decimal *to) { return val_decimal_from_date(to); }
+  String *val_str(String *to) { return val_string_from_date(to); }
+  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
+  {
+    bool res= m_value.to_TIME(current_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:
@@ -5084,11 +5189,13 @@ class Item_ref :public Item_ident,
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   String *val_str(String* tmp);
+  bool val_raw_native(String *str);
   bool is_null();
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
   double val_result();
   longlong val_int_result();
   String *str_result(String* tmp);
+  bool val_raw_native_result(String *str);
   my_decimal *val_decimal_result(my_decimal *);
   bool val_bool_result();
   bool is_null_result();
@@ -5293,6 +5400,7 @@ class Item_direct_ref :public Item_ref
   double val_real();
   longlong val_int();
   String *val_str(String* tmp);
+  bool val_raw_native(String *raw);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool is_null();
@@ -5389,6 +5497,7 @@ class Item_cache_wrapper :public Item_result_field,
   double val_real();
   longlong val_int();
   String *val_str(String* tmp);
+  bool val_raw_native(String *raw);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool is_null();
@@ -5585,6 +5694,12 @@ class Item_direct_view_ref :public Item_direct_ref
     else
       return Item_direct_ref::val_str(tmp);
   }
+  bool val_raw_native(String *raw)
+  {
+    if (check_null_ref())
+      return true;
+    return Item_direct_ref::val_raw_native(raw);
+  }
   my_decimal *val_decimal(my_decimal *tmp)
   {
     if (check_null_ref())
@@ -5730,6 +5845,7 @@ class Item_ref_null_helper: public Item_ref
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  bool val_raw_native(String *raw);
   virtual void print(String *str, enum_query_type query_type);
   table_map used_tables() const;
   Item *get_copy(THD *thd)
@@ -6592,6 +6708,27 @@ class Item_cache_date: public Item_cache_temporal
 };
 
 
+class Item_cache_timestamp: public Item_cache
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> m_raw;
+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();
+  longlong val_time_packed();
+  double val_real();
+  bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  int save_in_field(Field *field, bool no_conversions);
+  bool val_raw_native(String *str);
+};
+
+
 class Item_cache_real: public Item_cache
 {
   double value;
diff --git a/sql/item_cmpfunc.cc b/sql/item_cmpfunc.cc
index 4fa3cdc..f638b9c 100644
--- a/sql/item_cmpfunc.cc
+++ b/sql/item_cmpfunc.cc
@@ -568,6 +568,18 @@ bool Arg_comparator::set_cmp_func_datetime()
 }
 
 
+bool Arg_comparator::set_cmp_func_raw()
+{
+  THD *thd= current_thd;
+  m_compare_collation= &my_charset_numeric;
+  func= is_owner_equal_func() ? &Arg_comparator::compare_e_raw :
+                                &Arg_comparator::compare_raw;
+  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;
@@ -767,6 +779,33 @@ int Arg_comparator::compare_e_string()
 }
 
 
+int Arg_comparator::compare_raw()
+{
+  if (!(*a)->val_raw_with_conversion(&value1, compare_type_handler()))
+  {
+    if (!(*b)->val_raw_with_conversion(&value2, compare_type_handler()))
+    {
+      if (set_null)
+        owner->null_value= 0;
+      return compare_type_handler()->cmp_raw(&value1, &value2);
+    }
+  }
+  if (set_null)
+    owner->null_value= 1;
+  return -1;
+}
+
+
+int Arg_comparator::compare_e_raw()
+{
+  bool res1= (*a)->val_raw_with_conversion(&value1, compare_type_handler());
+  bool res2= (*b)->val_raw_with_conversion(&value2, compare_type_handler());
+  if (res1 || res2)
+    return MY_TEST(res1 == res2);
+  return MY_TEST(compare_type_handler()->cmp_raw(&value1, &value2) == 0);
+}
+
+
 int Arg_comparator::compare_real()
 {
   /*
@@ -2116,6 +2155,28 @@ longlong Item_func_between::val_int_cmp_time()
 }
 
 
+longlong Item_func_between::val_int_cmp_raw()
+{
+  const Type_handler *h= m_comparator.type_handler();
+  StringBuffer<STRING_BUFFER_USUAL_SIZE> value, a, b;
+  if (val_raw_with_conversion_from_item(args[0], &value, h))
+    return 0;
+  bool ra= args[1]->val_raw_with_conversion(&a, h);
+  bool rb= args[2]->val_raw_with_conversion(&b, h);
+  if (!ra && !rb)
+    return (longlong)
+      ((h->cmp_raw(&value, &a) >= 0 &&
+        h->cmp_raw(&value, &b) <= 0) != negated);
+  if (ra && rb)
+    null_value= true;
+  else if (ra)
+    null_value= h->cmp_raw(&value, &b) <= 0;
+  else
+    null_value= h->cmp_raw(&value ,&a) >= 0;
+  return (longlong) (!null_value && negated);
+}
+
+
 longlong Item_func_between::val_int_cmp_string()
 {
   String *value,*a,*b;
@@ -2295,6 +2356,15 @@ Item_func_ifnull::str_op(String *str)
 }
 
 
+bool Item_func_ifnull::raw_op(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (!val_raw_with_conversion_from_item(args[0], raw, type_handler()))
+    return false;
+  return val_raw_with_conversion_from_item(args[1], raw, type_handler());
+}
+
+
 bool Item_func_ifnull::date_op(MYSQL_TIME *ltime, ulonglong fuzzydate)
 {
   DBUG_ASSERT(fixed == 1);
@@ -2813,6 +2883,16 @@ Item_func_nullif::time_op(MYSQL_TIME *ltime)
 
 
 bool
+Item_func_nullif::raw_op(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (!compare())
+    return (null_value= true);
+  return val_raw_with_conversion_from_item(args[2], raw, type_handler());
+}
+
+
+bool
 Item_func_nullif::is_null()
 {
   return (null_value= (!compare() ? 1 : args[2]->is_null()));
@@ -2986,6 +3066,16 @@ bool Item_func_case::time_op(MYSQL_TIME *ltime)
 }
 
 
+bool Item_func_case::raw_op(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  Item *item= find_item();
+  if (!item)
+    return (null_value= true);
+  return val_raw_with_conversion_from_item(item, raw, type_handler());
+}
+
+
 bool Item_func_case::fix_fields(THD *thd, Item **ref)
 {
   bool res= Item_func::fix_fields(thd, ref);
@@ -3343,6 +3433,18 @@ bool Item_func_coalesce::time_op(MYSQL_TIME *ltime)
 }
 
 
+bool Item_func_coalesce::raw_op(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  for (uint i= 0; i < arg_count; i++)
+  {
+    if (!val_raw_with_conversion_from_item(args[i], raw, type_handler()))
+      return false;
+  }
+  return (null_value= true);
+}
+
+
 my_decimal *Item_func_coalesce::decimal_op(my_decimal *decimal_value)
 {
   DBUG_ASSERT(fixed == 1);
@@ -3620,6 +3722,54 @@ 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),
+   tmp("", 0)
+{}
+
+
+void in_timestamp::set(uint pos, Item *item)
+{
+  Timestamp_or_zero_datetime *buff= &((Timestamp_or_zero_datetime *) base)[pos];
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  if (item->val_raw_with_conversion(&raw, type_handler()))
+    buff->from_raw("", 0);
+  else
+    buff->from_raw(raw.ptr(), raw.length());
+}
+
+
+uchar *in_timestamp::get_value(Item *item)
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  if (item->val_raw_with_conversion(&raw, type_handler()))
+    return 0;
+  tmp.from_raw(raw.ptr(), raw.length());
+  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];
@@ -4028,6 +4178,48 @@ cmp_item *cmp_item_time::make_same()
 }
 
 
+void cmp_item_timestamp::store_value(Item *item)
+{
+  item->val_raw_with_conversion(&m_value, &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)
+{
+  String tmp;
+  arg->val_raw_with_conversion(&tmp, &type_handler_timestamp2);
+  return m_null_value || arg->null_value ? UNKNOWN :
+         type_handler_timestamp2.cmp_raw(&m_value, &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_raw(&m_value, &tmp->m_value);
+}
+
+
+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 08a462b..c6c50a6 100644
--- a/sql/item_cmpfunc.h
+++ b/sql/item_cmpfunc.h
@@ -89,6 +89,7 @@ class Arg_comparator: public Sql_alloc
   bool set_cmp_func_string();
   bool set_cmp_func_time();
   bool set_cmp_func_datetime();
+  bool set_cmp_func_raw();
   bool set_cmp_func_int();
   bool set_cmp_func_real();
   bool set_cmp_func_decimal();
@@ -121,6 +122,8 @@ class Arg_comparator: public Sql_alloc
   int compare_e_datetime();
   int compare_time();
   int compare_e_time();
+  int compare_raw();
+  int compare_e_raw();
   int compare_json_str_basic(Item *j, Item *s);
   int compare_json_str();
   int compare_str_json();
@@ -935,6 +938,7 @@ class Item_func_between :public Item_func_opt_neg
   longlong val_int_cmp_string();
   longlong val_int_cmp_datetime();
   longlong val_int_cmp_time();
+  longlong val_int_cmp_raw();
   longlong val_int_cmp_int();
   longlong val_int_cmp_real();
   longlong val_int_cmp_decimal();
@@ -1011,6 +1015,7 @@ class Item_func_coalesce :public Item_func_case_expression
   my_decimal *decimal_op(my_decimal *);
   bool date_op(MYSQL_TIME *ltime, ulonglong fuzzydate);
   bool time_op(MYSQL_TIME *ltime);
+  bool raw_op(String *raw);
   void fix_length_and_dec()
   {
     if (!aggregate_for_result(func_name(), args, arg_count, true))
@@ -1084,6 +1089,7 @@ class Item_func_ifnull :public Item_func_case_abbreviation2
   my_decimal *decimal_op(my_decimal *);
   bool date_op(MYSQL_TIME *ltime, ulonglong fuzzydate);
   bool time_op(MYSQL_TIME *ltime);
+  bool raw_op(String *raw);
   void fix_length_and_dec()
   {
     Item_func_case_abbreviation2::fix_length_and_dec2(args);
@@ -1140,6 +1146,10 @@ class Item_func_case_abbreviation2_switch: public Item_func_case_abbreviation2
   {
     return val_str_from_item(find_item(), str);
   }
+  bool raw_op(String *raw)
+  {
+    return val_raw_with_conversion_from_item(find_item(), raw, type_handler());
+  }
 };
 
 
@@ -1239,6 +1249,7 @@ class Item_func_nullif :public Item_func_case_expression
   longlong int_op();
   String *str_op(String *str);
   my_decimal *decimal_op(my_decimal *);
+  bool raw_op(String *raw);
   void fix_length_and_dec();
   bool walk(Item_processor processor, bool walk_subquery, void *arg);
   const char *func_name() const { return "nullif"; }
@@ -1402,6 +1413,19 @@ class in_longlong :public in_vector
 };
 
 
+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.
 */
@@ -1656,6 +1680,20 @@ class cmp_item_time: public cmp_item_temporal
   cmp_item *make_same();
 };
 
+
+class cmp_item_timestamp: public cmp_item_scalar
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> m_value;
+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;
@@ -2122,6 +2160,7 @@ class Item_func_case :public Item_func_case_expression
   my_decimal *decimal_op(my_decimal *);
   bool date_op(MYSQL_TIME *ltime, ulonglong fuzzydate);
   bool time_op(MYSQL_TIME *ltime);
+  bool raw_op(String *raw);
   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 ca61656..3707b47 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -2885,6 +2885,35 @@ my_decimal *Item_func_min_max::val_decimal_native(my_decimal *dec)
 }
 
 
+bool Item_func_min_max::val_raw_native(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  const Type_handler *handler= Item_hybrid_func::type_handler();
+  /*
+    For now we don't use val_raw_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_raw_native()
+      Item::val_raw_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 raw (instead of doing the operation in raw format).
+  */
+  StringBuffer<STRING_BUFFER_USUAL_SIZE> cur;
+  for (uint i= 0; i < arg_count; i++)
+  {
+    if (val_raw_with_conversion_from_item(args[i], i == 0 ? raw : &cur, handler))
+      return true;
+    if (i > 0)
+    {
+      int cmp= handler->cmp_raw(raw, &cur);
+      if ((cmp_sign < 0 ? cmp : -cmp) < 0 &&
+          raw->copy(cur.ptr(), cur.length(), &my_charset_bin))
+        return null_value= true;
+    }
+  }
+  return null_value= false;
+}
+
+
 longlong Item_func_bit_length::val_int()
 {
   DBUG_ASSERT(fixed == 1);
@@ -6651,6 +6680,14 @@ String *Item_func_last_value::val_str(String *str)
   return tmp;
 }
 
+
+bool Item_func_last_value::val_raw_native(String *raw)
+{
+  evaluate_sideeffects();
+  return val_raw_native_from_item(last_value, raw);
+}
+
+
 longlong Item_func_last_value::val_int()
 {
   longlong tmp;
diff --git a/sql/item_func.h b/sql/item_func.h
index 3b6cb4c..ad6ba50 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -534,6 +534,15 @@ class Item_func_hybrid_field_type: public Item_hybrid_func
   bool get_date_from_decimal_op(MYSQL_TIME *ltime, ulonglong fuzzydate);
   bool get_date_from_int_op(MYSQL_TIME *ltime, ulonglong fuzzydate);
 
+  /*
+    For now only the TIMESTAMP data type uses val_raw_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_raw_op()
+      get_date_from_raw_op()
+  */
+
 public:
   Item_func_hybrid_field_type(THD *thd):
     Item_hybrid_func(thd)
@@ -584,6 +593,24 @@ class Item_func_hybrid_field_type: public Item_hybrid_func
            Item_func_hybrid_field_type_get_date(this, res, fuzzy_date);
   }
 
+  bool val_raw_native(String *raw)
+  {
+    DBUG_ASSERT(fixed);
+    /*
+      For now only the TIMESTAMP data type uses val_raw_native().
+      Later we'll probably need new methods:
+        Item_func_hybrid_field_type::val_raw_native_from_{str|int|real|date}()
+        Type_handler::Item_func_hybrid_field_type_val_raw_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 raw format.
+      It may happen to be cheaper to do the operation in e.g. INT format and
+      then convert INT to raw (instead of doing the operation in raw format).
+      For now it's OK just to call raw_op() directly here.
+    */
+    return raw_op(raw);
+  }
+
   /**
      @brief Performs the operation that this functions implements when the
      result type is INT.
@@ -634,6 +661,7 @@ class Item_func_hybrid_field_type: public Item_hybrid_func
   */
   virtual bool time_op(MYSQL_TIME *res)= 0;
 
+  virtual bool raw_op(String *raw)= 0;
 };
 
 
@@ -701,6 +729,11 @@ class Item_func_numhybrid: public Item_func_hybrid_field_type
     DBUG_ASSERT(0);
     return true;
   }
+  bool raw_op(String *raw)
+  {
+    DBUG_ASSERT(0);
+    return true;
+  }
 };
 
 
@@ -1558,6 +1591,7 @@ class Item_func_min_max :public Item_hybrid_func
     return Item_func_min_max::type_handler()->
              Item_func_min_max_get_date(this, res, fuzzy_date);
   }
+  bool val_raw_native(String *raw);
   void aggregate_attributes_real(Item **items, uint nitems)
   {
     /*
@@ -1620,6 +1654,8 @@ class Item_func_rollup_const :public Item_func
   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_raw_native(String *raw)
+    { return val_raw_native_from_item(args[0], raw); }
   my_decimal *val_decimal(my_decimal *dec)
     { return val_decimal_from_item(args[0], dec); }
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate)
@@ -2904,6 +2940,13 @@ class Item_func_sp :public Item_func,
     return str;
   }
 
+  bool val_raw_native(String *raw)
+  {
+    if (execute())
+      return true;
+    return null_value= sp_result_field->val_raw_native(raw);
+  }
+
   void update_null_value()
   { 
     execute();
@@ -3040,6 +3083,7 @@ class Item_func_last_value :public Item_func
   String *val_str(String *);
   my_decimal *val_decimal(my_decimal *);
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
+  bool val_raw_native(String *);
   void 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 ca3316a..19cc982 100644
--- a/sql/item_subselect.cc
+++ b/sql/item_subselect.cc
@@ -1339,6 +1339,24 @@ String *Item_singlerow_subselect::val_str(String *str)
 }
 
 
+bool Item_singlerow_subselect::val_raw_native(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (forced_const)
+    return value->val_raw_native(raw);
+  if (!exec() && !value->null_value)
+  {
+    null_value= false;
+    return value->val_raw_native(raw);
+  }
+  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 5b27181..311184e 100644
--- a/sql/item_subselect.h
+++ b/sql/item_subselect.h
@@ -306,6 +306,7 @@ class Item_singlerow_subselect :public Item_subselect
   double val_real();
   longlong val_int ();
   String *val_str (String *);
+  bool val_raw_native(String *);
   my_decimal *val_decimal(my_decimal *);
   bool val_bool();
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
diff --git a/sql/item_sum.cc b/sql/item_sum.cc
index 69e228e..b779204 100644
--- a/sql/item_sum.cc
+++ b/sql/item_sum.cc
@@ -2378,6 +2378,15 @@ Item_sum_hybrid::val_str(String *str)
 }
 
 
+bool Item_sum_hybrid::val_raw_native(String *raw)
+{
+  DBUG_ASSERT(fixed == 1);
+  if (null_value)
+    return true;
+  return val_raw_native_from_item(value, raw);
+}
+
+
 void Item_sum_hybrid::cleanup()
 {
   DBUG_ENTER("Item_sum_hybrid::cleanup");
diff --git a/sql/item_sum.h b/sql/item_sum.h
index 89ebb04..1a93653 100644
--- a/sql/item_sum.h
+++ b/sql/item_sum.h
@@ -1069,6 +1069,7 @@ class Item_sum_hybrid :public Item_sum, public Type_handler_hybrid_field_type
   bool get_date(MYSQL_TIME *ltime, ulonglong fuzzydate);
   void reset_field();
   String *val_str(String *);
+  bool val_raw_native(String *);
   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 e01114f..ac5188a 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -1209,14 +1209,16 @@ bool Item_func_unix_timestamp::get_timestamp_value(my_time_t *seconds,
     }
   }
 
-  MYSQL_TIME ltime;
-  if (get_arg0_date(&ltime, TIME_NO_ZERO_IN_DATE))
-    return 1;
-
-  uint error_code;
-  *seconds= TIME_to_timestamp(current_thd, &ltime, &error_code);
-  *second_part= ltime.second_part;
-  return (null_value= (error_code == ER_WARN_DATA_OUT_OF_RANGE));
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  if (val_raw_with_conversion_from_item(args[0], &raw,
+                                        &type_handler_timestamp2))
+    return true;
+  Timestamp_or_zero_datetime tm(&raw);
+  if (tm.is_zero_datetime())
+    return null_value= true;
+  *seconds= tm.tm().tv_sec;
+  *second_part= tm.tm().tv_usec;
+  return false;
 }
 
 
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index 3607e6d..5810b94 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -130,6 +130,83 @@ bool Type_handler_data::init()
 Type_handler_data *type_handler_data= NULL;
 
 
+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;
+}
+
+
+void Timestamp::from_raw(const char *str, uint length)
+{
+  DBUG_ASSERT(length >= 4 && length <= 7);
+  uint dec= binary_length_to_precision(length);
+  my_timestamp_from_binary(this, (const uchar *) str, dec);
+}
+
+
+bool Timestamp::to_raw(String *raw, uint decimals) const
+{
+  uint len= my_timestamp_binary_length(decimals);
+  if (raw->reserve(len))
+    return true;
+  my_timestamp_to_binary(this, (uchar *) raw->ptr(), decimals);
+  raw->length(len);
+  return false;
+}
+
+
+bool Timestamp::to_TIME(THD *thd, MYSQL_TIME *to, ulonglong fuzzydate) const
+{
+  return thd->timestamp_to_TIME(to, tv_sec, tv_usec, fuzzydate);
+}
+
+
+Timestamp_or_zero_datetime::Timestamp_or_zero_datetime(THD *thd,
+                                                       const MYSQL_TIME *ltime,
+                                                       uint *error_code)
+{
+  tv_sec= TIME_to_timestamp(thd, ltime, error_code);
+  tv_usec= ltime->second_part;
+  if ((m_is_zero_datetime= (*error_code == ER_WARN_DATA_OUT_OF_RANGE)))
+  {
+    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,
+                                         ulonglong 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_raw(String *raw, uint decimals) const
+{
+  if (m_is_zero_datetime)
+  {
+    raw->length(0);
+    return false;
+  }
+  return Timestamp::to_raw(raw, decimals);
+}
+
+
 void Time::make_from_item(Item *item, const Options opt)
 {
   if (item->get_date(this, opt.get_date_flags()))
@@ -537,7 +614,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;
 }
 
 
@@ -722,6 +799,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
     {
@@ -805,7 +891,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:
@@ -2539,6 +2634,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
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  if (Item_val_raw_with_conversion(item, &raw))
+    return set_field_to_null_with_conversions(field, no_conversions);
+  field->set_notnull();
+  Timestamp_or_zero_datetime tm(&raw);
+  return tm.is_zero_datetime() ?
+         field->store_timestamp(0, 0) :
+         field->store_timestamp(tm.tm().tv_sec, tm.tm().tv_usec);
+}
+
+
 int Type_handler_string_result::Item_save_in_field(Item *item, Field *field,
                                                    bool no_conversions) const
 {
@@ -2605,6 +2716,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_raw();
+}
+
 
 /*************************************************************************/
 
@@ -2740,7 +2857,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 *
@@ -3751,6 +3868,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_raw();
+}
+
 longlong Type_handler_int_result::
            Item_func_between_val_int(Item_func_between *func) const
 {
@@ -3814,6 +3937,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)
@@ -3874,6 +4003,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
@@ -3975,6 +4113,17 @@ String *Type_handler_temporal_result::
 }
 
 
+String *Type_handler_timestamp_common::
+          Item_func_min_max_val_str(Item_func_min_max *func, String *str) const
+{
+  MYSQL_TIME ltime;
+  if (get_date_from_val_raw_native(func, &ltime) ||
+      my_TIME_to_str(&ltime, str, func->decimals))
+    return NULL;
+  return str;
+}
+
+
 String *Type_handler_int_result::
           Item_func_min_max_val_str(Item_func_min_max *func, String *str) const
 {
@@ -4013,6 +4162,16 @@ double Type_handler_temporal_result::
 }
 
 
+double Type_handler_timestamp_common::
+         Item_func_min_max_val_real(Item_func_min_max *func) const
+{
+  MYSQL_TIME ltime;
+  if (get_date_from_val_raw_native(func, &ltime))
+    return 0;
+  return TIME_to_double(&ltime);
+}
+
+
 double Type_handler_numeric::
          Item_func_min_max_val_real(Item_func_min_max *func) const
 {
@@ -4037,6 +4196,16 @@ longlong Type_handler_temporal_result::
 }
 
 
+longlong Type_handler_timestamp_common::
+         Item_func_min_max_val_int(Item_func_min_max *func) const
+{
+  MYSQL_TIME ltime;
+  if (get_date_from_val_raw_native(func, &ltime))
+    return 0;
+  return TIME_to_ulonglong(&ltime);
+}
+
+
 longlong Type_handler_numeric::
          Item_func_min_max_val_int(Item_func_min_max *func) const
 {
@@ -4071,6 +4240,17 @@ my_decimal *Type_handler_temporal_result::
 }
 
 
+my_decimal *Type_handler_timestamp_common::
+            Item_func_min_max_val_decimal(Item_func_min_max *func,
+                                          my_decimal *dec) const
+{
+  MYSQL_TIME ltime;
+  if (get_date_from_val_raw_native(func, &ltime))
+    return 0;
+  return date2my_decimal(&ltime, dec);
+}
+
+
 bool Type_handler_string_result::
        Item_func_min_max_get_date(Item_func_min_max *func,
                                   MYSQL_TIME *ltime, ulonglong fuzzydate) const
@@ -4107,6 +4287,13 @@ bool Type_handler_time_common::
   return func->get_time_native(ltime);
 }
 
+bool Type_handler_timestamp_common::
+       Item_func_min_max_get_date(Item_func_min_max *func,
+                                  MYSQL_TIME *ltime, ulonglong fuzzydate) const
+{
+  return get_date_from_val_raw_native(func, ltime);
+}
+
 /***************************************************************************/
 
 /**
@@ -5375,6 +5562,20 @@ bool Type_handler::
 }
 
 
+bool Type_handler::Item_send_timestamp(Item *item,
+                                       Protocol *protocol,
+                                       st_value *buf) const
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  MYSQL_TIME ltime;
+  if (item->val_raw_native(&raw))
+    return protocol->store_null();
+  return Timestamp_or_zero_datetime(&raw).to_TIME(current_thd, &ltime, 0) ?
+         protocol->store_null() :
+         protocol->store(&ltime, item->decimals);
+}
+
+
 bool Type_handler::
        Item_send_datetime(Item *item, Protocol *protocol, st_value *buf) const
 {
@@ -6507,6 +6708,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
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw0, raw1;
+  a->val_raw_with_conversion(&raw0, this);
+  b->val_raw_with_conversion(&raw1, this);
+  return !a->null_value && !b->null_value && !cmp_raw(&raw0, &raw1);
+}
+
+
 bool Type_handler_string_result::Item_eq_value(THD *thd,
                                                const Type_cmp_attributes *attr,
                                                Item *a, Item *b) const
@@ -6531,4 +6743,102 @@ bool Type_handler_decimal_result::Item_eq_value(THD *thd,
 
 }
 
+
 /***************************************************************************/
+
+bool Type_handler_timestamp_common::TIME_to_raw(THD *thd,
+                                                const MYSQL_TIME *ltime,
+                                                String *raw,
+                                                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_raw(raw, decimals);
+}
+
+
+bool
+Type_handler_timestamp_common::Item_val_raw_with_conversion(Item *item,
+                                                            String *raw)
+                                                            const
+{
+  MYSQL_TIME ltime;
+  if (item->type_handler()->is_timestamp_type())
+    return item->val_raw_native(raw);
+  return
+    item->get_date(&ltime, TIME_NO_ZERO_IN_DATE) ||
+    TIME_to_raw(current_thd, &ltime, raw, item->datetime_precision());
+}
+
+
+bool
+Type_handler_timestamp_common::Item_val_raw_with_conversion_result(Item *item,
+                                                                   String *raw)
+                                                                   const
+{
+  MYSQL_TIME ltime;
+  if (item->type_handler()->is_timestamp_type())
+    return item->val_raw_native_result(raw);
+  return
+    item->get_date_result(&ltime, TIME_NO_ZERO_IN_DATE) ||
+    TIME_to_raw(current_thd, &ltime, raw, item->datetime_precision());
+}
+
+
+int Type_handler_timestamp_common::cmp_raw(const String *a,
+                                           const String *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_raw_native(Item *item,
+                                                            MYSQL_TIME *ltime)
+                                                            const
+{
+  StringBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE> raw;
+  if (item->val_raw_native(&raw))
+  {
+    DBUG_ASSERT(item->null_value);
+    return true;
+  }
+  return item->null_value= Timestamp_or_zero_datetime(&raw).
+                             to_TIME(current_thd, ltime, 0);
+}
+
+
+bool
+Type_handler::Item_param_val_raw_native(Item_param *item, String *raw) const
+{
+  DBUG_ASSERT(0); // TODO-TYPE: MDEV-14271
+  return item->null_value= true;
+}
+
+
+bool
+Type_handler_timestamp_common::Item_param_val_raw_native(Item_param *item,
+                                                         String *raw) 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(&ltime, TIME_NO_ZERO_IN_DATE) ||
+    TIME_to_raw(current_thd, &ltime, raw, item->datetime_precision());
+}
diff --git a/sql/sql_type.h b/sql/sql_type.h
index f0ec0ba..07da914 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -77,6 +77,86 @@ struct SORT_FIELD_ATTR;
 class Vers_history_point;
 
 
+class Timestamp: protected timeval
+{
+  uint binary_length_to_precision(uint length);
+public:
+  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;
+  }
+  const struct timeval &tm() const { return *this; }
+  void set(const timeval &other) { timeval::operator=(other); }
+  void trunc(uint decimals) { my_timeval_trunc(this, decimals); }
+  bool to_TIME(THD *thd, MYSQL_TIME *ltime, ulonglong fuzzydate) const;
+  bool to_raw(String *to, uint decimals) const;
+  void from_raw(const char *str, uint length);
+};
+
+
+/*
+  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(const char *rawstr, uint length)
+  {
+    from_raw(rawstr, length);
+  }
+  Timestamp_or_zero_datetime(const String *raw)
+  {
+    from_raw(raw->ptr(), raw->length());
+  }
+  Timestamp_or_zero_datetime(bool is_zero_datetime, const struct timeval tm)
+  {
+    m_is_zero_datetime= is_zero_datetime;
+    Timestamp::set(tm);
+  }
+  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 &tm() const
+  {
+    DBUG_ASSERT(!is_zero_datetime());
+    return Timestamp::tm();
+  }
+  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, ulonglong fuzzydate) const;
+  /*
+    Convert to raw 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 raw values are binary comparable.
+  */
+  bool to_raw(String *to, uint decimals) const;
+  void from_raw(const char *str, uint length)
+  {
+    if (!(m_is_zero_datetime= (length == 0)))
+      Timestamp::from_raw(str, length);
+  }
+};
+
+
+
 /**
   Class Time is designed to store valid TIME values.
 
@@ -1052,6 +1132,7 @@ class Type_handler
   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)
@@ -1326,6 +1407,7 @@ class Type_handler
                                          Item_param *param,
                                          const Type_all_attributes *attr,
                                          const st_value *value) const= 0;
+  virtual bool Item_param_val_raw_native(Item_param *item, String *raw) 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;
@@ -1402,6 +1484,11 @@ class Type_handler
     DBUG_ASSERT(0);
     return NULL;
   }
+  virtual int cmp_raw(const String *a, const String *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
@@ -1426,6 +1513,16 @@ class Type_handler
   virtual
   bool Item_sum_variance_fix_length_and_dec(Item_sum_variance *) const= 0;
 
+  virtual bool Item_val_raw_with_conversion(Item *item, String *raw) const
+  {
+    return true;
+  }
+  virtual bool Item_val_raw_with_conversion_result(Item *item,
+                                                   String *raw) const
+  {
+    return true;
+  }
+
   virtual bool Item_val_bool(Item *item) const= 0;
   virtual bool Item_get_date(Item *item, MYSQL_TIME *ltime,
                              ulonglong fuzzydate) const= 0;
@@ -3165,6 +3262,9 @@ class Type_handler_datetime2: public Type_handler_datetime_common
 class Type_handler_timestamp_common: public Type_handler_temporal_with_date
 {
   static const Name m_name_timestamp;
+protected:
+  bool TIME_to_raw(THD *, const MYSQL_TIME *ltime, String *raw, uint dec) const;
+  bool get_date_from_val_raw_native(Item *item, MYSQL_TIME *ltime) const;
 public:
   virtual ~Type_handler_timestamp_common() {}
   const Name name() const { return m_name_timestamp; }
@@ -3178,6 +3278,20 @@ class Type_handler_timestamp_common: public Type_handler_temporal_with_date
   {
     return true;
   }
+  bool Item_eq_value(THD *thd, const Type_cmp_attributes *attr,
+                     Item *a, Item *b) const;
+  bool Item_val_raw_with_conversion(Item *item, String *raw) const;
+  bool Item_val_raw_with_conversion_result(Item *item, String *raw) const;
+  bool Item_param_val_raw_native(Item_param *item, String *raw) const;
+  int cmp_raw(const String *a, const String *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
   {
@@ -3190,10 +3304,12 @@ class Type_handler_timestamp_common: public Type_handler_temporal_with_date
   }
   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;
+  bool set_comparator_func(Arg_comparator *cmp) const;
   bool Item_hybrid_func_fix_attributes(THD *thd,
                                        const char *name,
                                        Type_handler_hybrid_field_type *,
@@ -3201,6 +3317,13 @@ class Type_handler_timestamp_common: public Type_handler_temporal_with_date
                                        Item **items, uint nitems) const;
   void Item_param_set_param_func(Item_param *param,
                                  uchar **pos, ulong len) const;
+  String *Item_func_min_max_val_str(Item_func_min_max *, String *) const;
+  double Item_func_min_max_val_real(Item_func_min_max *) const;
+  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 Item_func_min_max_get_date(Item_func_min_max*,
+                                  MYSQL_TIME *, ulonglong fuzzydate) const;
 };
 
 
