MDEV-13995 MAX(timestamp) returns a wrong result near DST change

26 files changed, 1850 insertions, 16 deletions

diff --git a/mysql-test/main/old-mode.result b/mysql-test/main/old-mode.result
index 73ad613048a..e0a3412bbdf 100644
--- a/mysql-test/main/old-mode.result
+++ b/mysql-test/main/old-mode.result
@@ -180,3 +180,44 @@ a	unix_timestamp(a)
 2010-10-31 02:25:26	1288481126
 drop table t1, t2;
 set time_zone=DEFAULT;
+#
+# MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+#
+SET global mysql56_temporal_format=false;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP(0));
+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, COALESCE(a), UNIX_TIMESTAMP(a) FROM t1;
+a	COALESCE(a)	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:26	2010-10-31 02:25:26	1288477526
+2010-10-31 02:25:25	2010-10-31 02:25:25	1288481125
+SELECT MIN(a), UNIX_TIMESTAMP(MIN(a)) AS a FROM t1;
+MIN(a)	a
+2010-10-31 02:25:26	1288477526
+SELECT MAX(a), UNIX_TIMESTAMP(MAX(a)) AS a FROM t1;
+MAX(a)	a
+2010-10-31 02:25:25	1288481125
+SELECT t1.a, UNIX_TIMESTAMP(t1.a), t2.a, UNIX_TIMESTAMP(t2.a) FROM t1 t1, t1 t2 WHERE t1.a=t2.a;
+a	UNIX_TIMESTAMP(t1.a)	a	UNIX_TIMESTAMP(t2.a)
+2010-10-31 02:25:26	1288477526	2010-10-31 02:25:26	1288477526
+2010-10-31 02:25:25	1288481125	2010-10-31 02:25:25	1288481125
+ALTER TABLE t1 MODIFY a TIMESTAMP(1);
+SELECT a, COALESCE(a), UNIX_TIMESTAMP(a) FROM t1;
+a	COALESCE(a)	UNIX_TIMESTAMP(a)
+2010-10-31 02:25:26.0	2010-10-31 02:25:26.0	1288477526.0
+2010-10-31 02:25:25.0	2010-10-31 02:25:25.0	1288481125.0
+SELECT MIN(a), UNIX_TIMESTAMP(MIN(a)) AS a FROM t1;
+MIN(a)	a
+2010-10-31 02:25:26.0	1288477526.0
+SELECT MAX(a), UNIX_TIMESTAMP(MAX(a)) AS a FROM t1;
+MAX(a)	a
+2010-10-31 02:25:25.0	1288481125.0
+SELECT t1.a, UNIX_TIMESTAMP(t1.a), t2.a, UNIX_TIMESTAMP(t2.a) FROM t1 t1, t1 t2 WHERE t1.a=t2.a;
+a	UNIX_TIMESTAMP(t1.a)	a	UNIX_TIMESTAMP(t2.a)
+2010-10-31 02:25:26.0	1288477526.0	2010-10-31 02:25:26.0	1288477526.0
+2010-10-31 02:25:25.0	1288481125.0	2010-10-31 02:25:25.0	1288481125.0
+DROP TABLE t1;
+SET time_zone=DEFAULT;
+SET global mysql56_temporal_format=true;
diff --git a/mysql-test/main/old-mode.test b/mysql-test/main/old-mode.test
index d7e8ce8ee55..7ec2092009b 100644
--- a/mysql-test/main/old-mode.test
+++ b/mysql-test/main/old-mode.test
@@ -119,3 +119,32 @@ insert t2 select a from t1;
 select a, unix_timestamp(a) from t2;
 drop table t1, t2;
 set time_zone=DEFAULT;
+
+
+--echo #
+--echo # MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+--echo #
+
+# This tests:
+#   Field_timestamp::val_native()
+#   Field_timestamp_hires::val_native()
+#   Type_handler_timestamp_common::type_handler_for_native_format()
+
+SET global mysql56_temporal_format=false;
+SET time_zone='+00:00';
+CREATE TABLE t1 (a TIMESTAMP(0));
+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, COALESCE(a), UNIX_TIMESTAMP(a) FROM t1;
+SELECT MIN(a), UNIX_TIMESTAMP(MIN(a)) AS a FROM t1;
+SELECT MAX(a), UNIX_TIMESTAMP(MAX(a)) AS a FROM t1;
+SELECT t1.a, UNIX_TIMESTAMP(t1.a), t2.a, UNIX_TIMESTAMP(t2.a) FROM t1 t1, t1 t2 WHERE t1.a=t2.a;
+ALTER TABLE t1 MODIFY a TIMESTAMP(1);
+SELECT a, COALESCE(a), UNIX_TIMESTAMP(a) FROM t1;
+SELECT MIN(a), UNIX_TIMESTAMP(MIN(a)) AS a FROM t1;
+SELECT MAX(a), UNIX_TIMESTAMP(MAX(a)) AS a FROM t1;
+SELECT t1.a, UNIX_TIMESTAMP(t1.a), t2.a, UNIX_TIMESTAMP(t2.a) FROM t1 t1, t1 t2 WHERE t1.a=t2.a;
+DROP TABLE t1;
+SET time_zone=DEFAULT;
+SET global mysql56_temporal_format=true;
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 ea1dc24386f..20f258ea261 100644
--- a/mysql-test/main/type_timestamp.result
+++ b/mysql-test/main/type_timestamp.result
@@ -1068,5 +1068,55 @@ DROP PROCEDURE p1;
 SET timestamp=DEFAULT;
 SET time_zone=DEFAULT;
 #
+# 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;
+#
+# Testing Type_handler_timestamp_common::Item_save_in_field()
+# "txt" is expected to have three fractional digits
+SET time_zone='+00:00';
+SET timestamp=UNIX_TIMESTAMP('2001-01-01 10:20:30.123456');
+CREATE TABLE t1 (ts1 TIMESTAMP(1) NOT NULL, ts2 TIMESTAMP(3) NOT NULL, txt TEXT);
+INSERT INTO t1 VALUES ('0000-00-00 00:00:00', '0000-00-00 00:00:00',COALESCE(ts1,ts2));
+INSERT INTO t1 VALUES (NOW(),NOW(),COALESCE(ts1,ts2));
+INSERT INTO t1 VALUES (NOW(1),NOW(3),COALESCE(ts1,ts2));
+SELECT * FROM t1;
+ts1	ts2	txt
+0000-00-00 00:00:00.0	0000-00-00 00:00:00.000	0000-00-00 00:00:00.000
+2001-01-01 10:20:30.0	2001-01-01 10:20:30.000	2001-01-01 10:20:30.000
+2001-01-01 10:20:30.1	2001-01-01 10:20:30.123	2001-01-01 10:20:30.100
+DROP TABLE t1;
+SET timestamp=DEFAULT;
+SET time_zone=DEFAULT;
+#
+# Testing Field_timestamp::store_native
+#
+SET sql_mode='';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES ('0000-00-00 00:00:00','0000-00-00 00:00:00');
+SET sql_mode='STRICT_ALL_TABLES,NO_ZERO_DATE';
+UPDATE t1 SET a=b;
+ERROR 22007: Incorrect datetime value: '0000-00-00 00:00:00' for column 'a' at row 1
+UPDATE t1 SET a=COALESCE(b);
+ERROR 22007: Incorrect datetime value: '0000-00-00 00:00:00' for column 'a' at row 1
+DROP TABLE t1;
+SET sql_mode=DEFAULT;
+#
 # End of 10.4 tests
 #
diff --git a/mysql-test/main/type_timestamp.test b/mysql-test/main/type_timestamp.test
index 033795ee183..28fa562cb53 100644
--- a/mysql-test/main/type_timestamp.test
+++ b/mysql-test/main/type_timestamp.test
@@ -661,5 +661,55 @@ SET timestamp=DEFAULT;
 SET time_zone=DEFAULT;
 
 --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 # Testing Type_handler_timestamp_common::Item_save_in_field()
+--echo # "txt" is expected to have three fractional digits
+SET time_zone='+00:00';
+SET timestamp=UNIX_TIMESTAMP('2001-01-01 10:20:30.123456');
+CREATE TABLE t1 (ts1 TIMESTAMP(1) NOT NULL, ts2 TIMESTAMP(3) NOT NULL, txt TEXT);
+INSERT INTO t1 VALUES ('0000-00-00 00:00:00', '0000-00-00 00:00:00',COALESCE(ts1,ts2));
+INSERT INTO t1 VALUES (NOW(),NOW(),COALESCE(ts1,ts2));
+INSERT INTO t1 VALUES (NOW(1),NOW(3),COALESCE(ts1,ts2));
+SELECT * FROM t1;
+DROP TABLE t1;
+SET timestamp=DEFAULT;
+SET time_zone=DEFAULT;
+
+--echo #
+--echo # Testing Field_timestamp::store_native
+--echo #
+
+SET sql_mode='';
+CREATE TABLE t1 (a TIMESTAMP, b TIMESTAMP);
+INSERT INTO t1 VALUES ('0000-00-00 00:00:00','0000-00-00 00:00:00');
+SET sql_mode='STRICT_ALL_TABLES,NO_ZERO_DATE';
+--error ER_TRUNCATED_WRONG_VALUE
+UPDATE t1 SET a=b;
+--error ER_TRUNCATED_WRONG_VALUE
+UPDATE t1 SET a=COALESCE(b);
+DROP TABLE t1;
+SET sql_mode=DEFAULT;
+
+
+--echo #
 --echo # End of 10.4 tests
 --echo #
diff --git a/mysql-test/main/type_timestamp_round.result b/mysql-test/main/type_timestamp_round.result
index 69b2c8fdeba..d6843a9fedf 100644
--- a/mysql-test/main/type_timestamp_round.result
+++ b/mysql-test/main/type_timestamp_round.result
@@ -162,3 +162,30 @@ SELECT * FROM t1,t2 WHERE COALESCE(t1.a)=t2.a;
 a	a
 20010101235959.9999999	2001-01-02 00:00:00
 DROP TABLE t1,t2;
+#
+# MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+#
+# Test Field_timestamp::store_native()
+#
+SET sql_mode=@default_sql_mode;
+SET time_zone='+00:00';
+CREATE TABLE t1 (ts0 TIMESTAMP, ts1 TIMESTAMP(1));
+INSERT INTO t1 VALUES ('2001-01-01 10:20:30', '2001-01-01 10:20:30.9');
+SELECT * FROM t1;
+ts0	ts1
+2001-01-01 10:20:30	2001-01-01 10:20:30.9
+# This should round
+UPDATE t1 SET ts0=COALESCE(ts1);
+SELECT * FROM t1;
+ts0	ts1
+2001-01-01 10:20:31	2001-01-01 10:20:30.9
+# Corner case
+UPDATE t1 SET ts1=FROM_UNIXTIME(2147483647.9);
+UPDATE t1 SET ts0=COALESCE(ts1);
+Warnings:
+Warning	1264	Out of range value for column 'ts0' at row 1
+SELECT * FROM t1;
+ts0	ts1
+2038-01-19 03:14:07	2038-01-19 03:14:07.9
+DROP TABLE t1;
+SET time_zone=DEFAULT;
diff --git a/mysql-test/main/type_timestamp_round.test b/mysql-test/main/type_timestamp_round.test
index 2ed01cc2a82..19e0ea86da5 100644
--- a/mysql-test/main/type_timestamp_round.test
+++ b/mysql-test/main/type_timestamp_round.test
@@ -136,3 +136,25 @@ INSERT INTO t2 VALUES ('2001-01-02 00:00:00');
 SELECT * FROM t1,t2 WHERE t1.a=t2.a;
 SELECT * FROM t1,t2 WHERE COALESCE(t1.a)=t2.a;
 DROP TABLE t1,t2;
+
+
+--echo #
+--echo # MDEV-13995 MAX(timestamp) returns a wrong result near DST change
+--echo #
+--echo # Test Field_timestamp::store_native()
+--echo #
+
+SET sql_mode=@default_sql_mode;
+SET time_zone='+00:00';
+CREATE TABLE t1 (ts0 TIMESTAMP, ts1 TIMESTAMP(1));
+INSERT INTO t1 VALUES ('2001-01-01 10:20:30', '2001-01-01 10:20:30.9');
+SELECT * FROM t1;
+--echo # This should round
+UPDATE t1 SET ts0=COALESCE(ts1);
+SELECT * FROM t1;
+--echo # Corner case
+UPDATE t1 SET ts1=FROM_UNIXTIME(2147483647.9);
+UPDATE t1 SET ts0=COALESCE(ts1);
+SELECT * FROM t1;
+DROP TABLE t1;
+SET time_zone=DEFAULT;
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 60ca9aa2426..1786c835bcd 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -5023,6 +5023,15 @@ my_time_t Field_timestamp::get_timestamp(const uchar *pos,
 }
 
 
+bool Field_timestamp::val_native(Native *to)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  my_time_t sec= (my_time_t) sint4korr(ptr);
+  return Timestamp_or_zero_datetime(Timestamp(sec, 0), sec == 0).
+           to_native(to, 0);
+}
+
+
 int Field_timestamp::store_TIME_with_warning(THD *thd, const Datetime *dt,
                                              const ErrConv *str, int was_cut)
 {
@@ -5143,6 +5152,14 @@ int Field_timestamp::store_timestamp_dec(const timeval &ts, uint dec)
   }
   if (ts.tv_sec == 0 && ts.tv_usec == 0 &&
       get_thd()->variables.sql_mode & (ulonglong) TIME_NO_ZERO_DATE)
+    return zero_time_stored_return_code_with_warning();
+  return 0;
+}
+
+
+int Field_timestamp::zero_time_stored_return_code_with_warning()
+{
+  if (get_thd()->variables.sql_mode & (ulonglong) TIME_NO_ZERO_DATE)
   {
     ErrConvString s(
       STRING_WITH_LEN("0000-00-00 00:00:00.000000") - (decimals() ? 6 - decimals() : 7),
@@ -5151,6 +5168,23 @@ int Field_timestamp::store_timestamp_dec(const timeval &ts, uint dec)
     return 1;
   }
   return 0;
+
+}
+
+
+int Field_timestamp::store_native(const Native &value)
+{
+  if (!value.length()) // Zero datetime
+  {
+    reset();
+    return zero_time_stored_return_code_with_warning();
+  }
+  /*
+    The exact second precision is not important here.
+    Field_timestamp*::store_timestamp_dec() do not use the "dec" parameter.
+    Passing TIME_SECOND_PART_DIGITS is OK.
+  */
+  return store_timestamp_dec(Timestamp(value).tv(), TIME_SECOND_PART_DIGITS);
 }
 
 
@@ -5410,6 +5444,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= (ulong) 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;
@@ -5516,6 +5562,19 @@ my_time_t Field_timestampf::get_timestamp(const uchar *pos,
 }
 
 
+bool Field_timestampf::val_native(Native *to)
+{
+  ASSERT_COLUMN_MARKED_FOR_READ;
+  // 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 Field_temporal::is_equal(Create_field *new_field)
 {
diff --git a/sql/field.h b/sql/field.h
index 19ab24bc2cb..f18ed9506f3 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -790,6 +790,15 @@ public:
     return store_timestamp_dec(Timeval(timestamp, sec_part),
                                TIME_SECOND_PART_DIGITS);
   }
+  /**
+    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,
@@ -836,6 +845,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
@@ -2735,6 +2749,7 @@ protected:
   {
     store_TIMEVAL(ts.tv());
   }
+  int zero_time_stored_return_code_with_warning();
 public:
   Field_timestamp(uchar *ptr_arg, uint32 len_arg,
                   uchar *null_ptr_arg, uchar null_bit_arg,
@@ -2785,6 +2800,8 @@ 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);
+  bool val_native(Native *to);
   uchar *pack(uchar *to, const uchar *from,
               uint max_length __attribute__((unused)))
   {
@@ -2864,6 +2881,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); }
@@ -2914,6 +2932,7 @@ public:
   {
     return get_timestamp(ptr, sec_part);
   }
+  bool val_native(Native *to);
   uint size_of() const { return sizeof(*this); }
 };
 
diff --git a/sql/filesort.cc b/sql/filesort.cc
index d0cb77ac63d..c76295e666a 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
+{
+  uint binlen= my_timestamp_binary_length(item->decimals);
+  Timestamp_or_zero_datetime_native_null native(current_thd, item);
+  if (native.is_null() || native.is_zero_datetime())
+  {
+    // 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..f86a27d587a 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,62 @@ Item *Item_cache_time::make_literal(THD *thd)
   return new (thd->mem_root) Item_time_literal(thd, &ltime, decimals);
 }
 
+
+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);
+  return m_native.save_in_field(field, 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);
+}
+
+
+Datetime Item_cache_timestamp::to_datetime(THD *thd)
+{
+  DBUG_ASSERT(is_fixed() == 1);
+  if (!has_value())
+  {
+    null_value= true;
+    return Datetime();
+  }
+  return Datetime(thd, Timestamp_or_zero_datetime(m_native).tv());
+}
+
+
+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;
+}
+
+
 bool Item_cache_real::cache_value()
 {
   if (!example)
diff --git a/sql/item.h b/sql/item.h
index 2ee2eabddac..5d0f2cd43f5 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,54 @@ 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; }
+  int save_in_field(Field *field, bool no_conversions)
+  {
+    Timestamp_or_zero_datetime_native native(m_value, decimals);
+    return native.save_in_field(field, decimals);
+  }
+  longlong val_int()
+  {
+    return m_value.to_datetime(current_thd).to_longlong();
+  }
+  double val_real()
+  {
+    return m_value.to_datetime(current_thd).to_double();
+  }
+  String *val_str(String *to)
+  {
+    return m_value.to_datetime(current_thd).to_string(to, decimals);
+  }
+  my_decimal *val_decimal(my_decimal *to)
+  {
+    return m_value.to_datetime(current_thd).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;
+  }
+  bool val_native(THD *thd, Native *to)
+  {
+    return m_value.to_native(to, decimals);
+  }
+  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 +5191,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 +5402,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 +5499,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 +5696,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 +5847,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 +6671,48 @@ public:
 };
 
 
+class Item_cache_timestamp: public Item_cache
+{
+  Timestamp_or_zero_datetime_native m_native;
+  Datetime to_datetime(THD *thd);
+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 *to)
+  {
+    return to_datetime(current_thd).to_string(to, decimals);
+  }
+  my_decimal *val_decimal(my_decimal *to)
+  {
+    return to_datetime(current_thd).to_decimal(to);
+  }
+  longlong val_int()
+  {
+    return to_datetime(current_thd).to_longlong();
+  }
+  double val_real()
+  {
+    return to_datetime(current_thd).to_double();
+  }
+  longlong val_datetime_packed(THD *thd)
+  {
+    DBUG_ASSERT(0);
+    return 0;
+  }
+  longlong val_time_packed(THD *thd)
+  {
+    DBUG_ASSERT(0);
+    return 0;
+  }
+  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..4cd63042558 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_or_zero_datetime_native_null native(current_thd, item, true);
+  if (native.is_null())
+    *buff= Timestamp_or_zero_datetime();
+  else
+    *buff= Timestamp_or_zero_datetime(native);
+}
+
+
+uchar *in_timestamp::get_value(Item *item)
+{
+  Timestamp_or_zero_datetime_native_null native(current_thd, item, true);
+  if (native.is_null())
+    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,49 @@ 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)
+{
+  THD *thd= current_thd;
+  Timestamp_or_zero_datetime_native_null tmp(thd, arg, true);
+  return m_null_value || tmp.is_null() ? 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..882f92465b1 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_or_zero_datetime_native 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..1f949a4445b 100644
--- a/sql/item_func.cc
+++ b/sql/item_func.cc
@@ -2730,6 +2730,28 @@ 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();
+  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 +6475,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..36589bf2b2e 100644
--- a/sql/item_func.h
+++ b/sql/item_func.h
@@ -768,6 +768,12 @@ public:
            Item_func_hybrid_field_type_get_date_with_warn(thd, this, to, mode);
   }
 
+  bool val_native(THD *thd, Native *to)
+  {
+    DBUG_ASSERT(fixed);
+    return native_op(thd, to);
+  }
+
   /**
      @brief Performs the operation that this functions implements when the
      result type is INT.
@@ -838,6 +844,7 @@ public:
   */
   virtual bool time_op(THD *thd, MYSQL_TIME *res)= 0;
 
+  virtual bool native_op(THD *thd, Native *native)= 0;
 };
 
 
@@ -905,6 +912,11 @@ public:
     DBUG_ASSERT(0);
     return true;
   }
+  bool native_op(THD *thd, Native *to)
+  {
+    DBUG_ASSERT(0);
+    return true;
+  }
 };
 
 
@@ -1771,6 +1783,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 +1847,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 +3168,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 +3304,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..4b072e253c5 100644
--- a/sql/item_timefunc.cc
+++ b/sql/item_timefunc.cc
@@ -1217,15 +1217,13 @@ 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_or_zero_datetime_native_null native(current_thd, args[0], true);
+  if ((null_value= native.is_null() || native.is_zero_datetime()))
     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);
+  *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..e415aacce75 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -209,6 +209,104 @@ 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);
+}
+
+
+int Timestamp_or_zero_datetime_native::save_in_field(Field *field,
+                                                     uint decimals) const
+{
+  field->set_notnull();
+  if (field->type_handler()->type_handler_for_native_format() ==
+      &type_handler_timestamp2)
+    return field->store_native(*this);
+  if (is_zero_datetime())
+  {
+    static Datetime zero(Datetime::zero());
+    return field->store_time_dec(zero.get_mysql_time(), decimals);
+  }
+  return field->store_timestamp_dec(Timestamp(*this).tv(), decimals);
+}
+
+
 void Sec6::make_from_decimal(const my_decimal *d, ulong *nanoseconds)
 {
   m_neg= my_decimal2seconds(d, &m_sec, &m_usec, nanoseconds);
@@ -1272,7 +1370,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;
 }
 
 
@@ -1283,6 +1381,15 @@ const Type_handler *Type_handler_row::type_handler_for_comparison() const
 
 /***************************************************************************/
 
+const Type_handler *
+Type_handler_timestamp_common::type_handler_for_native_format() const
+{
+  return &type_handler_timestamp2;
+}
+
+
+/***************************************************************************/
+
 const Type_handler *Type_handler_typelib::type_handler_for_item_field() const
 {
   return &type_handler_string;
@@ -1457,6 +1564,16 @@ 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->type_handler_for_native_format() ==
+          &type_handler_timestamp2)
+        m_type_handler= &type_handler_datetime;
     }
     else
     {
@@ -1540,7 +1657,19 @@ 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->type_handler_for_native_format() ==
+         &type_handler_timestamp2) +
+        (h->type_handler_for_native_format() ==
+         &type_handler_timestamp2) == 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 +3524,18 @@ 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_or_zero_datetime_native_null tmp(field->table->in_use, item, true);
+  if (tmp.is_null())
+    return set_field_to_null_with_conversions(field, no_conversions);
+  return tmp.save_in_field(field, item->decimals);
+}
+
+
 int Type_handler_string_result::Item_save_in_field(Item *item, Field *field,
                                                    bool no_conversions) const
 {
@@ -3461,6 +3602,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 +3743,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 +4910,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 +4979,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 +5045,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 +5172,9 @@ 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);
+  THD *thd= current_thd;
+  return Timestamp_or_zero_datetime_native_null(thd, func).
+           to_datetime(thd).to_string(str, func->decimals);
 }
 
 
@@ -5056,10 +5226,13 @@ 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();
+  THD *thd= current_thd;
+  return Timestamp_or_zero_datetime_native_null(thd, func).
+           to_datetime(thd).to_double();
 }
 
 
@@ -5101,7 +5274,9 @@ 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();
+  THD *thd= current_thd;
+  return Timestamp_or_zero_datetime_native_null(thd, func).
+           to_datetime(thd).to_longlong();
 }
 
 
@@ -5156,7 +5331,9 @@ 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);
+  THD *thd= current_thd;
+  return Timestamp_or_zero_datetime_native_null(thd, func).
+           to_datetime(thd).to_decimal(dec);
 }
 
 
@@ -5208,6 +5385,15 @@ 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 Timestamp_or_zero_datetime_native_null(thd, func).
+           to_datetime(thd).copy_to_mysql_time(ltime);
+}
+
 /***************************************************************************/
 
 /**
@@ -6480,6 +6666,18 @@ bool Type_handler::
 }
 
 
+bool Type_handler::Item_send_timestamp(Item *item,
+                                       Protocol *protocol,
+                                       st_value *buf) const
+{
+  Timestamp_or_zero_datetime_native_null native(protocol->thd, item);
+  if (native.is_null())
+    return protocol->store_null();
+  native.to_TIME(protocol->thd, &buf->value.m_time);
+  return protocol->store(&buf->value.m_time, item->decimals);
+}
+
+
 bool Type_handler::
        Item_send_datetime(Item *item, Protocol *protocol, st_value *buf) const
 {
@@ -7628,6 +7826,16 @@ 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_or_zero_datetime_native_null na(thd, a, true);
+  Timestamp_or_zero_datetime_native_null nb(thd, b, true);
+  return !na.is_null() && !nb.is_null() && !cmp_native(na, nb);
+}
+
+
 bool Type_handler_string_result::Item_eq_value(THD *thd,
                                                const Type_cmp_attributes *attr,
                                                Item *a, Item *b) const
@@ -7845,3 +8053,113 @@ 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()->type_handler_for_native_format() ==
+      &type_handler_timestamp2)
+    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()->type_handler_for_native_format() ==
+      &type_handler_timestamp2)
+    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));
+}
+
+
+Timestamp_or_zero_datetime_native_null::
+  Timestamp_or_zero_datetime_native_null(THD *thd, Item *item, bool conv)
+   :Null_flag(false)
+{
+  DBUG_ASSERT(item->type_handler()->type_handler_for_native_format() ==
+              &type_handler_timestamp2 || conv);
+  if (conv ?
+      type_handler_timestamp2.Item_val_native_with_conversion(thd, item, this) :
+      item->val_native(thd, this))
+    Null_flag::operator=(true);
+  // If no conversion, then is_null() should be equal to item->null_value
+  DBUG_ASSERT(is_null() == item->null_value || conv);
+  /*
+    is_null() can be true together with item->null_value==false, which means
+    a non-NULL item was evaluated, but then the conversion to TIMESTAMP failed.
+    But is_null() can never be false if item->null_value==true.
+  */
+  DBUG_ASSERT(is_null() >= item->null_value);
+}
+
+
+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 0aa142d5c37..925dc25d7e7 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:
@@ -1925,7 +1945,16 @@ public:
     { }
   };
 
+  static Datetime zero()
+  {
+    int warn;
+    static Longlong_hybrid nr(0, false);
+    return Datetime(&warn, nr, date_mode_t(0));
+  }
 public:
+  Datetime() // NULL value
+   :Temporal_with_date()
+  { }
   Datetime(THD *thd, Item *item, date_mode_t fuzzydate)
    :Temporal_with_date(thd, item, fuzzydate)
   {
@@ -2204,6 +2233,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)
@@ -2237,7 +2267,18 @@ public:
   explicit Timestamp(const 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);
@@ -2273,6 +2314,117 @@ 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);
+  Datetime to_datetime(THD *thd) const
+  {
+    return Datetime(thd, *this);
+  }
+  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;
+};
+
+
+/**
+  A helper class to store non-null MariaDB TIMESTAMP values in
+  the native binary encoded representation.
+*/
+class Timestamp_or_zero_datetime_native:
+          public NativeBuffer<STRING_BUFFER_TIMESTAMP_BINARY_SIZE>
+{
+public:
+  Timestamp_or_zero_datetime_native() { }
+  Timestamp_or_zero_datetime_native(const Timestamp_or_zero_datetime &ts,
+                                    uint decimals)
+  {
+    if (ts.to_native(this, decimals))
+      length(0); // safety
+  }
+  int save_in_field(Field *field, uint decimals) const;
+  bool is_zero_datetime() const
+  {
+    return length() == 0;
+  }
+};
+
+
+/**
+  A helper class to store nullable MariaDB TIMESTAMP values in
+  the native binary encoded representation.
+*/
+class Timestamp_or_zero_datetime_native_null: public Timestamp_or_zero_datetime_native,
+                                              public Null_flag
+{
+public:
+  // With optional data type conversion
+  Timestamp_or_zero_datetime_native_null(THD *thd, Item *item, bool conv);
+  // Without data type conversion: item is known to be of the TIMESTAMP type
+  Timestamp_or_zero_datetime_native_null(THD *thd, Item *item)
+   :Timestamp_or_zero_datetime_native_null(thd, item, false)
+  { }
+  Datetime to_datetime(THD *thd) const
+  {
+    return is_null() ? Datetime() :
+                       Datetime(thd, Timestamp_or_zero_datetime(*this).tv());
+  }
+  void to_TIME(THD *thd, MYSQL_TIME *to)
+  {
+    DBUG_ASSERT(!is_null());
+    Datetime::Options opt(TIME_CONV_NONE, TIME_FRAC_NONE);
+    Timestamp_or_zero_datetime(*this).to_TIME(thd, to, opt);
+  }
+  bool is_zero_datetime() const
+  {
+    DBUG_ASSERT(!is_null());
+    return Timestamp_or_zero_datetime_native::is_zero_datetime();
+  }
+};
+
+
 /*
   Flags for collation aggregation modes, used in TDCollation::agg():
 
@@ -2905,6 +3057,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)
@@ -3026,6 +3179,10 @@ public:
   */
   virtual bool is_param_long_data_type() const { return false; }
   virtual const Type_handler *type_handler_for_comparison() const= 0;
+  virtual const Type_handler *type_handler_for_native_format() const
+  {
+    return this;
+  }
   virtual const Type_handler *type_handler_for_item_field() const
   {
     return this;
@@ -3208,6 +3365,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;
@@ -3310,6 +3470,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
@@ -3334,6 +3499,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,
@@ -5187,10 +5363,13 @@ 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;
 public:
   virtual ~Type_handler_timestamp_common() {}
   const Name name() const { return m_name_timestamp; }
   const Type_handler *type_handler_for_comparison() const;
+  const Type_handler *type_handler_for_native_format() const;
   enum_field_types field_type() const { return MYSQL_TYPE_TIMESTAMP; }
   enum_mysql_timestamp_type mysql_timestamp_type() const
   {
@@ -5201,6 +5380,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
   {
@@ -5213,8 +5406,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;
@@ -5222,6 +5416,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 *,
@@ -5229,6 +5424,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 c7af86ee6a5..8728dee1918 100644
--- a/sql/structs.h
+++ b/sql/structs.h
@@ -847,6 +847,8 @@ public:
 
 class Timeval: public timeval
 {
+protected:
+  Timeval() { }
 public:
   Timeval(my_time_t sec, ulong usec)
   {