MDEV-16983 Application-time periods: foreign key PART 1/3

This work adds support for FOREIGN KEYS for SQL:2011 Application-time periods feature.
* Two hidden keys are added in key matching a referred foreign key;
* Referential integrity check is done completely on sql layer, in handler;
* A new struct FOREIGN_KEY is introduced to store a prelocked referential metadata in TABLE

Limitations:
* Only versionde-to-versioned and plain-to-plain references are supported at this moment
* innodb: a flag denoting a period is added to FK flags,
  see dict_create_add_foreign_to_dictionary() and dict_load_foreign().
  In this way only one period per table is supported, and it's quite poorly scalable.

In this commit:
 * Update is treated as delete + insert which causes false-positives. (will bi fixed in PART 2)
 * Self-referencing tables have false-negatives during deletes, since the same row could be
 fetched during lookup (will be fixed in PART 3)

28 files changed, 1269 insertions, 121 deletions

diff --git a/mysql-test/suite/period/r/fk.result b/mysql-test/suite/period/r/fk.result
new file mode 100644
index 00000000000..82c77c794ab
--- /dev/null
+++ b/mysql-test/suite/period/r/fk.result
@@ -0,0 +1,260 @@
+set default_storage_engine= innodb;
+create or replace table t (id int, x int, s date, e date, period for p(s,e),
+unique(id, x, p without overlaps));
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+foreign key(id, x, period fp)
+references t(id, x, period p)
+on delete restrict);
+show create table s;
+Table	Create Table
+s	CREATE TABLE `s` (
+  `id` int(11) NOT NULL,
+  `x` int(11) NOT NULL,
+  `s` date NOT NULL,
+  `e` date NOT NULL,
+  PERIOD FOR `fp` (`s`, `e`),
+  KEY `id` (`id`,`x`,`s`,`e`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`id`, `x`, `s`, `e`) REFERENCES `t` (`id`, `x`, `s`, `e`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+flush tables;
+Warnings:
+Note	1105	Foreign table test.s is not opened
+insert into s values(1, 1, '2017-01-03', '2017-01-20');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+insert into t values(1, 1, '2017-01-20', '2017-01-25');
+insert into t values(1, 1, '2017-01-25', '2017-01-26');
+insert into t values(1, 1, '2017-01-26', '2017-01-30');
+insert into s values(1, 1, '2017-01-03', '2017-01-20');
+select * from t;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-20	2017-01-25
+1	1	2017-01-25	2017-01-26
+1	1	2017-01-26	2017-01-30
+select * from s;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+delete from t;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+select * from t;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-20	2017-01-25
+1	1	2017-01-25	2017-01-26
+1	1	2017-01-26	2017-01-30
+select * from s;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+delete from t where s = '2017-01-03' and e = '2017-01-20';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# no error
+delete from t where s = '2017-01-20' and e = '2017-01-25';
+insert into t values(1, 1, '2017-01-20', '2017-01-25');
+insert into s values (1, 1, '2017-01-27', '2017-01-30');
+delete from t where s = '2017-01-26' and e = '2017-01-30';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+delete from t where s = '2017-01-25' and e = '2017-01-26';
+insert into s values (1, 1, '2017-01-22', '2017-01-28');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+insert into t values (1, 1, '2017-01-25', '2017-01-26');
+select * from t;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-20	2017-01-25
+1	1	2017-01-25	2017-01-26
+1	1	2017-01-26	2017-01-30
+select * from s;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-27	2017-01-30
+insert into s values (1, 1, '2017-01-03', '2017-01-15');
+insert into s values (1, 1, '2017-01-07', '2017-01-15');
+insert into s values (1, 1, '2017-01-07', '2017-01-20');
+insert into s values (1, 1, '2017-01-07', '2017-01-26');
+insert into s values (1, 1, '2017-01-07', '2017-01-27');
+insert into s values (1, 1, '2017-01-01', '2017-02-28');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+insert into s values (1, 1, '2017-01-01', '2017-01-30');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+select * from t;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-20	2017-01-25
+1	1	2017-01-25	2017-01-26
+1	1	2017-01-26	2017-01-30
+select * from s;
+id	x	s	e
+1	1	2017-01-03	2017-01-15
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-07	2017-01-15
+1	1	2017-01-07	2017-01-20
+1	1	2017-01-07	2017-01-26
+1	1	2017-01-07	2017-01-27
+1	1	2017-01-27	2017-01-30
+update t set x= 2 where s='2017-01-03' and e='2017-01-20';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+update s set x= 2 where s = '2017-01-03' and e = '2017-01-20';
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+update s set s= '2017-01-05' where s < '2017-01-05' and e > '2017-01-05';
+update s set s= '2017-01-01' where s < '2017-01-26' and e > '2017-01-25';
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+# Free period ('2017-01-25', '2017-01-26') from references
+update s set s= '2017-01-26', e= '2017-01-30' where s < '2017-01-26'
+                                                    and e > '2017-01-25';
+update t set x= 2 where s = '2017-01-25' and e = '2017-01-26';
+update t set s= '2017-01-26', e= '2017-01-30' where s = '2017-01-25'
+                                                    and e = '2017-01-26';
+update s set x= 2 where s = '2017-01-26' and e = '2017-01-30';
+update s set s= '2017-01-28', e = '2017-01-29' where x = 2;
+select * from t;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+1	1	2017-01-20	2017-01-25
+1	1	2017-01-26	2017-01-30
+1	2	2017-01-26	2017-01-30
+select * from s;
+id	x	s	e
+1	1	2017-01-05	2017-01-15
+1	1	2017-01-05	2017-01-20
+1	1	2017-01-07	2017-01-15
+1	1	2017-01-07	2017-01-20
+1	1	2017-01-27	2017-01-30
+1	2	2017-01-28	2017-01-29
+1	2	2017-01-28	2017-01-29
+update t set x= 2 where s='2017-01-03' and e='2017-01-20';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+#
+# False-positives
+#
+# Expand left
+update t set s= '2017-01-01' where s = '2017-01-03' and e = '2017-01-20';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# Shrink left
+update t set s= '2017-01-05' where e = '2017-01-20';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# Expand right
+update t set e= '2017-02-10' where s = '2017-01-26' and e = '2017-01-30';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# Shrink right
+update t set e= '2017-01-29' where s = '2017-01-26';
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+delete from s where s = '2017-01-27' and e = '2017-01-30';
+update t set e= '2017-01-29' where s = '2017-01-26' and x = 1;
+# Shrink both
+# Not a false-positive
+update t set s= '2017-01-27', e= '2017-01-28' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# False-positive
+update t set s= '2017-01-28', e= '2017-01-29' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# Expand both
+update t set s= '2017-01-20', e= '2017-02-05' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# Move right
+# Not a false-positive
+update t set s= '2017-02-02', e= '2017-02-25' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# False-positive
+update t set s= '2017-01-28', e= '2017-02-25' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# Move left
+# Not a false-positive
+update t set s= '2017-01-20', e= '2017-01-27' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+# False-positive
+update t set s= '2017-01-20', e= '2017-01-29' where x = 2;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+drop table s;
+delete from t;
+create or replace table s (x int, y int, z char(200), pk int,
+e date, s date, period for fp(s,e),
+unique(x), unique(z),
+foreign key(pk, y, period fp)
+references t(id, x, period p)
+on delete restrict);
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+delete from t;
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+show create table s;
+Table	Create Table
+s	CREATE TABLE `s` (
+  `x` int(11) DEFAULT NULL,
+  `y` int(11) NOT NULL,
+  `z` char(200) DEFAULT NULL,
+  `pk` int(11) NOT NULL,
+  `e` date NOT NULL,
+  `s` date NOT NULL,
+  PERIOD FOR `fp` (`s`, `e`),
+  UNIQUE KEY `x` (`x`),
+  UNIQUE KEY `z` (`z`),
+  KEY `pk` (`pk`,`y`,`s`,`e`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`pk`, `y`, `s`, `e`) REFERENCES `t` (`id`, `x`, `s`, `e`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+select * from t;
+id	x	s	e
+1	1	2017-01-03	2017-01-20
+insert into s values (-1, 1, '0', 1, '2017-01-15', '2017-01-05');
+delete from t;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+delete from s;
+delete from t;
+drop table s;
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+foreign key(id, x, period no_such_p)
+references t(id, x, period p)
+on delete restrict);
+ERROR HY000: Period `no_such_p` is not found in table
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+foreign key(id, x, s, e)
+references t(id, x, period p)
+on delete restrict);
+ERROR 42000: Incorrect foreign key definition for 'foreign key without name': Key reference and table reference don't match
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+foreign key(id, x, period fp)
+references t(id, x, s, e)
+on delete restrict);
+ERROR 42000: Incorrect foreign key definition for 'foreign key without name': Key reference and table reference don't match
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+foreign key(id, x, period fp)
+references t(id, x, period no_such_p)
+on delete restrict);
+ERROR HY000: Period `no_such_p` is not found in referenced table `test`.`t`
+create or replace table t1 (id int, x int, s timestamp, e timestamp, period for p(s,e),
+unique(id, x, p without overlaps));
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+foreign key(id, x, period fp)
+references t1(id, x, period p)
+on delete restrict);
+ERROR HY000: Fields of `fp` and `test`.`t1`.`p` have different types
+create or replace table s (id int, x int, s timestamp(6), e timestamp(6),
+period for fp(s,e),
+foreign key(id, x, period fp)
+references t1(id, x, period p)
+on delete restrict);
+ERROR HY000: Fields of `fp` and `test`.`t1`.`p` have different types
+create or replace table s (id int, x int, s date, e date);
+alter table s add period for fp(s,e),
+add foreign key(id, x, period fp)
+references t1(id, x, period no_such_p)
+on delete restrict;
+ERROR HY000: Period `no_such_p` is not found in referenced table `test`.`t1`
+alter table s add period for fp(s,e),
+add foreign key(id, x, period fp)
+references t1(id, x, period p)
+on delete restrict;
+ERROR HY000: Fields of `fp` and `test`.`t1`.`p` have different types
+alter table s change s s timestamp(6), change e e timestamp(6),
+add period for fp(s, e),
+add foreign key(id, x, period fp)
+references t1(id, x, period p)
+on delete restrict;
+ERROR HY000: Fields of `fp` and `test`.`t1`.`p` have different types
+alter table s change s s timestamp(6), change e e timestamp(6),
+add period for fp(s, e);
+alter table s add foreign key(id, x, period fp)
+references t1(id, x, period p)
+on delete restrict;
+ERROR HY000: Fields of `fp` and `test`.`t1`.`p` have different types
+drop database test;
+create database test;
diff --git a/mysql-test/suite/period/t/fk.test b/mysql-test/suite/period/t/fk.test
new file mode 100644
index 00000000000..1168a1b185b
--- /dev/null
+++ b/mysql-test/suite/period/t/fk.test
@@ -0,0 +1,239 @@
+--source include/have_innodb.inc
+set default_storage_engine= innodb;
+
+create or replace table t (id int, x int, s date, e date, period for p(s,e),
+                           unique(id, x, p without overlaps));
+
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+                           foreign key(id, x, period fp)
+                             references t(id, x, period p)
+                             on delete restrict);
+
+show create table s;
+
+flush tables;
+
+--error ER_NO_REFERENCED_ROW_2
+insert into s values(1, 1, '2017-01-03', '2017-01-20');
+
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+insert into t values(1, 1, '2017-01-20', '2017-01-25');
+insert into t values(1, 1, '2017-01-25', '2017-01-26');
+insert into t values(1, 1, '2017-01-26', '2017-01-30');
+
+insert into s values(1, 1, '2017-01-03', '2017-01-20');
+select * from t;
+select * from s;
+
+--error ER_ROW_IS_REFERENCED_2
+delete from t;
+
+select * from t;
+select * from s;
+
+--error ER_ROW_IS_REFERENCED_2
+delete from t where s = '2017-01-03' and e = '2017-01-20';
+
+--echo # no error
+delete from t where s = '2017-01-20' and e = '2017-01-25';
+insert into t values(1, 1, '2017-01-20', '2017-01-25');
+
+insert into s values (1, 1, '2017-01-27', '2017-01-30');
+--error ER_ROW_IS_REFERENCED_2
+delete from t where s = '2017-01-26' and e = '2017-01-30';
+
+delete from t where s = '2017-01-25' and e = '2017-01-26';
+--error ER_NO_REFERENCED_ROW_2
+insert into s values (1, 1, '2017-01-22', '2017-01-28');
+insert into t values (1, 1, '2017-01-25', '2017-01-26');
+
+select * from t;
+select * from s;
+
+insert into s values (1, 1, '2017-01-03', '2017-01-15');
+insert into s values (1, 1, '2017-01-07', '2017-01-15');
+insert into s values (1, 1, '2017-01-07', '2017-01-20');
+insert into s values (1, 1, '2017-01-07', '2017-01-26');
+insert into s values (1, 1, '2017-01-07', '2017-01-27');
+--error ER_NO_REFERENCED_ROW_2
+insert into s values (1, 1, '2017-01-01', '2017-02-28');
+--error ER_NO_REFERENCED_ROW_2
+insert into s values (1, 1, '2017-01-01', '2017-01-30');
+
+select * from t;
+select * from s;
+
+--error ER_ROW_IS_REFERENCED_2
+update t set x= 2 where s='2017-01-03' and e='2017-01-20';
+
+--error ER_NO_REFERENCED_ROW_2
+update s set x= 2 where s = '2017-01-03' and e = '2017-01-20';
+
+update s set s= '2017-01-05' where s < '2017-01-05' and e > '2017-01-05';
+
+--error ER_NO_REFERENCED_ROW_2
+update s set s= '2017-01-01' where s < '2017-01-26' and e > '2017-01-25';
+
+--echo # Free period ('2017-01-25', '2017-01-26') from references
+update s set s= '2017-01-26', e= '2017-01-30' where s < '2017-01-26'
+                                                    and e > '2017-01-25';
+
+
+update t set x= 2 where s = '2017-01-25' and e = '2017-01-26';
+update t set s= '2017-01-26', e= '2017-01-30' where s = '2017-01-25'
+                                                    and e = '2017-01-26';
+
+update s set x= 2 where s = '2017-01-26' and e = '2017-01-30';
+update s set s= '2017-01-28', e = '2017-01-29' where x = 2;
+
+select * from t;
+select * from s;
+
+--error ER_ROW_IS_REFERENCED_2
+update t set x= 2 where s='2017-01-03' and e='2017-01-20';
+
+
+--echo #
+--echo # False-positives
+--echo #
+
+--echo # Expand left
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-01' where s = '2017-01-03' and e = '2017-01-20';
+
+--echo # Shrink left
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-05' where e = '2017-01-20';
+
+--echo # Expand right
+--error ER_ROW_IS_REFERENCED_2
+update t set e= '2017-02-10' where s = '2017-01-26' and e = '2017-01-30';
+
+--echo # Shrink right
+--error ER_ROW_IS_REFERENCED_2
+update t set e= '2017-01-29' where s = '2017-01-26';
+
+delete from s where s = '2017-01-27' and e = '2017-01-30';
+update t set e= '2017-01-29' where s = '2017-01-26' and x = 1;
+
+--echo # Shrink both
+--echo # Not a false-positive
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-27', e= '2017-01-28' where x = 2;
+--echo # False-positive
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-28', e= '2017-01-29' where x = 2;
+
+--echo # Expand both
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-20', e= '2017-02-05' where x = 2;
+
+--echo # Move right
+--echo # Not a false-positive
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-02-02', e= '2017-02-25' where x = 2;
+--echo # False-positive
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-28', e= '2017-02-25' where x = 2;
+
+--echo # Move left
+--echo # Not a false-positive
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-20', e= '2017-01-27' where x = 2;
+--echo # False-positive
+--error ER_ROW_IS_REFERENCED_2
+update t set s= '2017-01-20', e= '2017-01-29' where x = 2;
+
+# TODO remove when deadlock issue will be solved
+drop table s;
+delete from t;
+create or replace table s (x int, y int, z char(200), pk int,
+                           e date, s date, period for fp(s,e),
+                           unique(x), unique(z),
+                           foreign key(pk, y, period fp)
+                             references t(id, x, period p)
+                             on delete restrict);
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+delete from t;
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+
+show create table s;
+select * from t;
+insert into s values (-1, 1, '0', 1, '2017-01-15', '2017-01-05');
+#--error ER_NO_REFERENCED_ROW_2
+#insert into s values ( 0, 1, '1', 1, '2017-02-20', '2017-02-03');
+--error ER_ROW_IS_REFERENCED_2
+delete from t;
+delete from s;
+delete from t;
+
+# TODO remove when deadlock issue will be solved
+drop table s;
+--error ER_PERIOD_NOT_FOUND
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+                           foreign key(id, x, period no_such_p)
+                             references t(id, x, period p)
+                             on delete restrict);
+--error ER_WRONG_FK_DEF
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+                           foreign key(id, x, s, e)
+                             references t(id, x, period p)
+                             on delete restrict);
+--error ER_WRONG_FK_DEF
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+                           foreign key(id, x, period fp)
+                             references t(id, x, s, e)
+                             on delete restrict);
+
+--error ER_PERIOD_FK_NOT_FOUND
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+                           foreign key(id, x, period fp)
+                             references t(id, x, period no_such_p)
+                             on delete restrict);
+
+
+create or replace table t1 (id int, x int, s timestamp, e timestamp, period for p(s,e),
+                           unique(id, x, p without overlaps));
+
+--error ER_PERIOD_FK_TYPES_MISMATCH
+create or replace table s (id int, x int, s date, e date, period for fp(s,e),
+                           foreign key(id, x, period fp)
+                             references t1(id, x, period p)
+                             on delete restrict);
+
+--error ER_PERIOD_FK_TYPES_MISMATCH
+create or replace table s (id int, x int, s timestamp(6), e timestamp(6),
+                           period for fp(s,e),
+                           foreign key(id, x, period fp)
+                             references t1(id, x, period p)
+                             on delete restrict);
+
+create or replace table s (id int, x int, s date, e date);
+--error ER_PERIOD_FK_NOT_FOUND
+alter table s add period for fp(s,e),
+              add foreign key(id, x, period fp)
+               references t1(id, x, period no_such_p)
+               on delete restrict;
+
+--error ER_PERIOD_FK_TYPES_MISMATCH
+alter table s add period for fp(s,e),
+              add foreign key(id, x, period fp)
+               references t1(id, x, period p)
+               on delete restrict;
+
+--error ER_PERIOD_FK_TYPES_MISMATCH
+alter table s change s s timestamp(6), change e e timestamp(6),
+              add period for fp(s, e),
+              add foreign key(id, x, period fp)
+               references t1(id, x, period p)
+               on delete restrict;
+
+alter table s change s s timestamp(6), change e e timestamp(6),
+              add period for fp(s, e);
+--error ER_PERIOD_FK_TYPES_MISMATCH
+alter table s add foreign key(id, x, period fp)
+               references t1(id, x, period p)
+               on delete restrict;
+
+drop database test;
+create database test;
diff --git a/sql/field.cc b/sql/field.cc
index de2a1b2bb4c..7076d1f04ae 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -4473,17 +4473,17 @@ void Field_longlong::set_max()
   int8store(ptr, unsigned_flag ? ULONGLONG_MAX : LONGLONG_MAX);
 }
 
-bool Field_longlong::is_max()
+bool Field_longlong::is_max_in_ptr(const uchar *ptr_arg) const
 {
   DBUG_ASSERT(marked_for_read());
   if (unsigned_flag)
   {
     ulonglong j;
-    j= uint8korr(ptr);
+    j= uint8korr(ptr_arg);
     return j == ULONGLONG_MAX;
   }
   longlong j;
-  j= sint8korr(ptr);
+  j= sint8korr(ptr_arg);
   return j == LONGLONG_MAX;
 }
 
@@ -5555,13 +5555,13 @@ void Field_timestampf::set_max()
   DBUG_VOID_RETURN;
 }
 
-bool Field_timestampf::is_max()
+bool Field_timestampf::is_max_in_ptr(const uchar *ptr_arg) const
 {
-  DBUG_ENTER("Field_timestampf::is_max");
+  DBUG_ENTER("Field_timestampf::is_max_in_ptr");
   DBUG_ASSERT(marked_for_read());
 
-  DBUG_RETURN(mi_sint4korr(ptr) == TIMESTAMP_MAX_VALUE &&
-              mi_sint3korr(ptr + 4) == TIME_MAX_SECOND_PART);
+  DBUG_RETURN(mi_sint4korr(ptr_arg) == TIMESTAMP_MAX_VALUE &&
+              mi_sint3korr(ptr_arg + 4) == TIME_MAX_SECOND_PART);
 }
 
 my_time_t Field_timestampf::get_timestamp(const uchar *pos,
@@ -7492,7 +7492,8 @@ uint Field_string::max_packed_col_length(uint max_length)
 }
 
 
-uint Field_string::get_key_image(uchar *buff, uint length, const uchar *ptr_arg, imagetype type_arg) const
+uint Field_string::get_key_image(uchar *buff, uint length,
+                                 const uchar *ptr_arg, imagetype type_arg) const
 {
   size_t bytes= my_charpos(field_charset(), (char*) ptr_arg,
                            (char*) ptr_arg + field_length,
diff --git a/sql/field.h b/sql/field.h
index 925a6d5146e..4f81bd5f87d 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -771,8 +771,13 @@ public:
    */
   virtual void set_max()
   { DBUG_ASSERT(0); }
-  virtual bool is_max()
+  virtual bool is_max_in_ptr(const uchar *ptr_arg) const
   { DBUG_ASSERT(0); return false; }
+  bool is_max(const uchar *record) const
+  {
+    return is_max_in_ptr(ptr_in_record(record));
+  }
+  bool is_max() const { return is_max_in_ptr(ptr); }
 
   uchar		*ptr;			// Position to field in record
 
@@ -1477,7 +1482,8 @@ public:
       Number of copied bytes (excluding padded zero bytes -- see above).
   */
 
-  virtual uint get_key_image(uchar *buff, uint length, const uchar *ptr_arg, imagetype type_arg) const
+  virtual uint get_key_image(uchar *buff, uint length,
+                             const uchar *ptr_arg, imagetype type_arg) const
   {
     get_image(buff, length, ptr_arg, &my_charset_bin);
     return length;
@@ -2686,7 +2692,7 @@ public:
     return unpack_int64(to, from, from_end);
   }
   void set_max() override;
-  bool is_max() override;
+  bool is_max_in_ptr(const uchar *ptr_arg) const override;
   ulonglong get_max_int_value() const override
   {
     return unsigned_flag ? 0xFFFFFFFFFFFFFFFFULL : 0x7FFFFFFFFFFFFFFFULL;
@@ -3225,7 +3231,7 @@ public:
     return memcmp(a_ptr, b_ptr, pack_length());
   }
   void set_max() override;
-  bool is_max() override;
+  bool is_max_in_ptr(const uchar *ptr_arg) const override;
   my_time_t get_timestamp(const uchar *pos, ulong *sec_part) const override;
   my_time_t get_timestamp(ulong *sec_part) const
   {
@@ -3872,7 +3878,7 @@ public:
   bool has_charset() const override { return charset() != &my_charset_bin; }
   Field *make_new_field(MEM_ROOT *root, TABLE *new_table, bool keep_type)
     override;
-  uint get_key_image(uchar *buff, uint length, 
+  uint get_key_image(uchar *buff, uint length,
                      const uchar *ptr_arg, imagetype type) const override;
   sql_mode_t value_depends_on_sql_mode() const override;
   sql_mode_t can_handle_sql_mode_dependency_on_store() const override;
diff --git a/sql/handler.cc b/sql/handler.cc
index eeabc40d940..d6194fa5fb1 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -6672,6 +6672,10 @@ int handler::ha_write_row(const uchar *buf)
   mark_trx_read_write();
   increment_statistics(&SSV::ha_write_count);
 
+  error= period_row_ins_fk_check(buf);
+  if (unlikely(error))
+    DBUG_RETURN(error);
+
   if (table->s->long_unique_table)
   {
     if (this->inited == RND)
@@ -6728,6 +6732,9 @@ int handler::ha_update_row(const uchar *old_data, const uchar *new_data)
   {
     return error;
   }
+  error= period_row_upd_fk_check(old_data, new_data);
+  if (error)
+    return error;
 
   TABLE_IO_WAIT(tracker, m_psi, PSI_TABLE_UPDATE_ROW, active_index, 0,
                       { error= update_row(old_data, new_data);})
@@ -6776,6 +6783,272 @@ int handler::update_first_row(const uchar *new_data)
 }
 
 
+static int period_find_overlapping_record(handler *handler,
+                                          const uchar *key_buf,
+                                          const uchar *record_to_cmp,
+                                          uchar *record,
+                                          const KEY &key_to_cmp,
+                                          const KEY &key)
+{
+  // number of key parts not including period fields
+  auto base_parts= key.user_defined_key_parts - 2;
+
+  /*
+     We should evaluate SELECT start, end WHERE start < @end and end > @start.
+     one is < and one is >, so we can not do it by O(1) queries to btree.
+     You can think about spatial tree, but don't forget about base_parts
+     to check! So it's actually WHERE idx1 = @idx1 and idx2 = @idx2,...
+
+     What we can do here is to apply merge optimization, to decide whether is
+     better to use spatial tree (when available), or btree.
+
+     Or we can use a btree with an rtree in leaves, if it will ever be
+     implemented.
+
+     TODO check WITHOUT OVERLAPS on key since it is only chance to have it O(1).
+   */
+  int error= handler->ha_index_read_map(record, key_buf,
+                                        PREV_BITS(ulong, base_parts),
+                                        HA_READ_KEY_OR_NEXT);
+  if (unlikely(error))
+      return error;
+
+  while (key_period_compare_bases(key, key_to_cmp, record, record_to_cmp) == 0)
+  {
+    int overlap= key_period_compare_periods(key, key_to_cmp,
+                                            record, record_to_cmp);
+
+    if (overlap > 0) // all the rest will definitely succeed tested record
+      return HA_ERR_KEY_NOT_FOUND;
+    else if (overlap == 0)
+      return 0;
+
+    error= handler->ha_index_next(record);
+    if (unlikely(error == HA_ERR_END_OF_FILE))
+      return HA_ERR_KEY_NOT_FOUND;
+    else if (unlikely(error))
+      return error;
+  }
+
+  return HA_ERR_KEY_NOT_FOUND;
+}
+
+
+static
+void set_bits_with_key(MY_BITMAP *map, const KEY *key, uint key_parts)
+{
+  for (uint i = 0; i < key_parts; i++)
+    bitmap_set_bit(map, key->key_part[i].fieldnr - 1);
+}
+
+/**
+@param record                      record to update from, or a deleted record */
+int handler::period_row_del_fk_check(const uchar *record)
+{
+  if (!table->referenced_keys)
+    return 0;
+  if (table->versioned() && !table->vers_end_field()->is_max(record))
+    return 0;
+
+  for(int k= 0; k < table->referenced_keys; k++)
+  {
+    auto &fk= table->referenced[k];
+    if (!fk.has_period)
+      continue;
+
+    DBUG_ASSERT(fk.fields_num == fk.foreign_key->user_defined_key_parts);
+    DBUG_ASSERT(fk.fields_num == fk.referenced_key->user_defined_key_parts);
+
+    handler *foreign_handler= fk.foreign_key->table->file;
+
+    /* We shouldn't save cursor here, since this handler is never used.
+       The foreign table is only opened for FK matches. */
+    int error= foreign_handler->ha_index_init(fk.foreign_key_nr, false);
+    if(error)
+      return error;
+
+
+    set_bits_with_key(fk.foreign_key->table->read_set,
+                      fk.foreign_key, fk.fields_num);
+    auto *record_buffer= foreign_handler->get_table()->record[0];
+    auto *key_buffer= foreign_handler->get_table()->record[1];
+
+    key_copy(key_buffer, record, fk.referenced_key, 0);
+    error= period_find_overlapping_record(foreign_handler,
+                                          key_buffer,
+                                          record,
+                                          record_buffer,
+                                          *fk.referenced_key,
+                                          *fk.foreign_key);
+
+    int end_error= foreign_handler->ha_index_end();
+
+    if (error == HA_ERR_KEY_NOT_FOUND) // no key matched
+      return end_error;
+    if (error)
+      return error;
+    if (end_error)
+      return end_error;
+    if (key_period_compare_periods(*fk.referenced_key,
+                                   *fk.foreign_key,
+                                   record, record_buffer) == 0)
+      return HA_ERR_ROW_IS_REFERENCED;
+  }
+  return 0;
+}
+
+/**
+   A lightweight wrapper around memcmp for equal-sized buffers
+ */
+class Binary_value
+{
+  uchar *ptr;
+  uint size;
+  int cmp(const Binary_value &rhs) const
+  {
+    DBUG_ASSERT(rhs.size == size);
+    return memcmp(ptr, rhs.ptr, size);
+  }
+public:
+  Binary_value(uchar *ptr, uint size): ptr(ptr), size(size) {}
+  bool operator < (const Binary_value &rhs) const { return cmp(rhs) < 0; }
+  bool operator > (const Binary_value &rhs) const { return cmp(rhs) > 0; }
+  bool operator == (const Binary_value &rhs) const { return cmp(rhs) == 0; }
+  bool operator != (const Binary_value &rhs) const { return cmp(rhs) != 0; }
+  bool operator <= (const Binary_value &rhs) const { return cmp(rhs) <= 0; }
+  bool operator >= (const Binary_value &rhs) const { return cmp(rhs) >= 0; }
+  void fill(const uchar *rhs)
+  {
+    memcpy(ptr, rhs, size);
+  }
+  void fill(const Binary_value &val)
+  {
+    fill(val.ptr);
+  }
+  Binary_value &operator = (const Binary_value &) = delete;
+};
+
+/*
+ @return  0                      All the range from record is covered by ref_key
+          HA_ERR_KEY_NOT_FOUND   Some part of the range is not covered
+          other value            Handler returned error, and it's not
+                                 HA_ERR_KEY_NOT_FOUND and not HA_ERR_END_OF_FILE
+ */
+static int period_check_row_references(handler *ref_handler,
+                                       const uchar *key_buf,
+                                       const uchar *record,
+                                       uchar *ref_record,
+                                       const KEY &key,
+                                       const KEY &ref_key)
+{
+  int error= period_find_overlapping_record(ref_handler, key_buf,
+                                            record, ref_record,
+                                            key, ref_key);
+  if (error)
+    return error;
+
+  auto period_start= key.user_defined_key_parts - 2;
+  auto period_end= key.user_defined_key_parts - 1;
+
+  auto *foreign_start_field= key.key_part[period_start].field;
+  auto *foreign_end_field= key.key_part[period_end].field;
+  auto *ref_start_field= ref_key.key_part[period_start].field;
+  auto *ref_end_field= ref_key.key_part[period_end].field;
+
+  uchar values[4][Type_handler_datetime::hires_bytes(MAX_DATETIME_PRECISION)];
+  auto field_len= key.key_part[period_start].field->pack_length();
+
+  Binary_value foreign_start(values[0], field_len);
+  Binary_value foreign_end(values[1], field_len);
+  Binary_value last_start(values[2], field_len);
+  Binary_value last_end(values[3], field_len);
+
+  foreign_start.fill(foreign_start_field->ptr_in_record(record));
+  foreign_end.fill(foreign_end_field->ptr_in_record(record));
+  last_start.fill(ref_start_field->ptr_in_record(ref_record));
+  last_end.fill(ref_end_field->ptr_in_record(ref_record));
+
+  // leftmost referenced record is to the right from foreign record
+  if (foreign_start < last_start)
+    return HA_ERR_KEY_NOT_FOUND;
+
+  while (last_end < foreign_end)
+  {
+    error= ref_handler->ha_index_next(ref_record);
+
+    if (unlikely(error == HA_ERR_END_OF_FILE))
+      return HA_ERR_KEY_NOT_FOUND;
+    else if (unlikely(error))
+      return error;
+
+    if (key_period_compare_periods(key, ref_key, record, ref_record) != 0)
+      return HA_ERR_KEY_NOT_FOUND;
+
+    last_start.fill(ref_start_field->ptr_in_record(ref_record));
+    if (last_end != last_start)
+      return HA_ERR_KEY_NOT_FOUND;
+    last_end.fill(ref_end_field->ptr_in_record(ref_record));
+  }
+  DBUG_ASSERT(!error);
+  return 0;
+}
+
+
+int handler::period_row_ins_fk_check(const uchar *record)
+{
+  if (!table->foreign_keys)
+    return 0;
+
+  for(int k= 0; k < table->foreign_keys; k++)
+  {
+    auto &fk= table->foreign[k];
+    if (!fk.has_period)
+      continue;
+    if (table->versioned() && !table->vers_end_field()->is_max(record))
+      return 0;
+
+    DBUG_ASSERT(fk.fields_num == fk.foreign_key->user_defined_key_parts);
+    DBUG_ASSERT(fk.fields_num == fk.referenced_key->user_defined_key_parts);
+
+    handler *ref_handler= fk.referenced_key->table->file;
+
+    int error= ref_handler->ha_index_init(fk.referenced_key_nr, false);
+    if(error)
+      return error;
+
+    set_bits_with_key(fk.referenced_key->table->read_set,
+                      fk.referenced_key, fk.fields_num);
+
+    auto *ref_record= ref_handler->get_table()->record[0];
+    auto *key_buffer= ref_handler->get_table()->record[1];
+
+    key_copy(key_buffer, record, fk.foreign_key, 0);
+    bool row_references;
+    error= period_check_row_references(ref_handler, key_buffer,
+                                       record, ref_record,
+                                       *fk.foreign_key, *fk.referenced_key);
+
+    int end_error= ref_handler->ha_index_end();
+
+    if (error == HA_ERR_KEY_NOT_FOUND)
+      return HA_ERR_NO_REFERENCED_ROW;
+    else if (error)
+      return error;
+    else if (end_error)
+      return end_error;
+  }
+  return 0;
+}
+
+
+int handler::period_row_upd_fk_check(const uchar *old_data, const uchar *new_data)
+{
+  int error= period_row_del_fk_check(old_data);
+  if (!error)
+    error= period_row_ins_fk_check(new_data);
+  return error;
+}
+
 int handler::ha_delete_row(const uchar *buf)
 {
   int error;
@@ -6788,6 +7061,10 @@ int handler::ha_delete_row(const uchar *buf)
   DBUG_ASSERT(buf == table->record[0] ||
               buf == table->record[1]);
 
+  error= period_row_del_fk_check(buf);
+  if (unlikely(error))
+    return error;
+
   MYSQL_DELETE_ROW_START(table_share->db.str, table_share->table_name.str);
   mark_trx_read_write();
   increment_statistics(&SSV::ha_delete_count);
diff --git a/sql/handler.h b/sql/handler.h
index b90c3c57ef0..ee41a6950ad 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -4590,6 +4590,10 @@ private:
     return HA_ERR_WRONG_COMMAND;
   }
 
+  int period_row_del_fk_check(const uchar *record);
+  int period_row_ins_fk_check(const uchar *record);
+  int period_row_upd_fk_check(const uchar *old_data, const uchar *new_data);
+
   /* Perform initialization for a direct update request */
 public:
   int ha_direct_update_rows(ha_rows *update_rows, ha_rows *found_rows);
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index d8038be4320..bf7a84cc8cb 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7949,3 +7949,7 @@ ER_PERIOD_WITHOUT_OVERLAPS_PARTITIONED
         eng "Period WITHOUT OVERLAPS is not implemented for partitioned tables"
 ER_PERIOD_WITHOUT_OVERLAPS_NON_UNIQUE
         eng "Period WITHOUT OVERLAPS is only allowed for unique keys"
+ER_PERIOD_FK_TYPES_MISMATCH
+        eng "Fields of %`s and %`s.%`s.%`s have different types"
+ER_PERIOD_FK_NOT_FOUND
+        eng "Period %`s is not found in referenced table %`s.%`s"
\ No newline at end of file
diff --git a/sql/sql_base.cc b/sql/sql_base.cc
index 64434acd9e7..1d7dc173bf0 100644
--- a/sql/sql_base.cc
+++ b/sql/sql_base.cc
@@ -3541,6 +3541,12 @@ open_and_process_routine(THD *thd, Query_tables_list *prelocking_ctx,
   DBUG_RETURN(FALSE);
 }
 
+static bool table_can_need_prelocking(THD *thd, TABLE_LIST *table)
+{
+  return (table->updating && table->lock_type >= TL_WRITE_ALLOW_WRITE)
+         || thd->lex->default_used;
+}
+
 /*
   If we are not already in prelocked mode and extended table list is not
   yet built we might have to build the prelocking set for this statement.
@@ -3555,12 +3561,9 @@ bool extend_table_list(THD *thd, TABLE_LIST *tables,
 {
   bool error= false;
   LEX *lex= thd->lex;
-  bool maybe_need_prelocking=
-    (tables->updating && tables->lock_type >= TL_WRITE_ALLOW_WRITE)
-    || thd->lex->default_used;
 
   if (thd->locked_tables_mode <= LTM_LOCK_TABLES &&
-      ! has_prelocking_list && maybe_need_prelocking)
+      ! has_prelocking_list && table_can_need_prelocking(thd, tables))
   {
     bool need_prelocking= FALSE;
     TABLE_LIST **save_query_tables_last= lex->query_tables_last;
@@ -4165,6 +4168,82 @@ open_tables_check_upgradable_mdl(THD *thd, TABLE_LIST *tables_start,
   return FALSE;
 }
 
+static int find_index_by_fields(const TABLE *table,
+                                const List<LEX_CSTRING> &names)
+{
+
+  for (uint k= 0; k < table->s->keys; k++)
+  {
+    auto &key= table->key_info[k];
+    if (names.elements > key.user_defined_key_parts)
+      continue;
+
+    bool match= true;
+    uint kp= 0;
+    for(const auto &name: names)
+    {
+      if (!Lex_ident(name).streq(key.key_part[kp].field->field_name))
+      {
+        match= false;
+        break;
+      }
+      kp++;
+    }
+
+    if (match)
+      return k;
+  }
+  return -1;
+}
+
+TABLE *find_fk_open_table(THD *thd, const char *db, size_t db_len,
+                          const char *table, size_t table_len);
+
+
+static
+FOREIGN_KEY *convert_foreign_key_list(THD *thd,
+                                      TABLE *foreign_table_init,
+                                      TABLE *referenced_table_init,
+                                      const List<FOREIGN_KEY_INFO> &fk_list,
+                                      MEM_ROOT *table_root)
+{
+  auto *fk_buf= (FOREIGN_KEY*)
+                alloc_root(table_root, fk_list.elements * sizeof(FOREIGN_KEY));
+  auto * const fk_buf_result= fk_buf;
+
+  for(const auto &fk: fk_list)
+  {
+    TABLE *foreign_table = foreign_table_init
+                           ? foreign_table_init
+                           : find_fk_open_table(thd, fk.foreign_db->str,
+                                                fk.foreign_db->length,
+                                                fk.foreign_table->str,
+                                                fk.foreign_table->length);
+    TABLE *referenced_table = referenced_table_init
+                              ? referenced_table_init
+                              : find_fk_open_table(thd, fk.referenced_db->str,
+                                                   fk.referenced_db->length,
+                                                   fk.referenced_table->str,
+                                                   fk.referenced_table->length);
+    DBUG_ASSERT(referenced_table != NULL && foreign_table != NULL);
+
+    int fk_no= find_index_by_fields(foreign_table, fk.foreign_fields);
+    int ref_no= find_index_by_fields(referenced_table, fk.referenced_fields);
+
+    DBUG_ASSERT(fk_no >= 0 && ref_no >= 0);
+
+    // Emplace the new object in fk_buf
+    new(fk_buf) FOREIGN_KEY {uint(fk_no), uint(ref_no),
+                             &foreign_table->key_info[fk_no],
+                             &referenced_table->key_info[ref_no],
+                             fk.referenced_fields.elements,
+                             fk.has_period,
+                             fk.delete_method,
+                             fk.update_method};
+    fk_buf++;
+  }
+  return fk_buf_result;
+}
 
 /**
   Open all tables in list
@@ -4466,6 +4545,49 @@ restart:
     }
   }
 
+  /* After all the tables are opened, including prelocked foreign key
+     child and parent tables, we can fill out the foreign and referential info.
+   */
+  for (tables= *start; tables; tables= tables->next_global)
+  {
+    /*
+       FK-prelocked tables are ignored -- anyway, they'll miss their parents or
+       childs.
+       When FK handling will be moved to sql, we'll perhaps need to prelock
+       referential tables recursively when CASCADE option is specified, and
+       then we'll need to fill FK-prelocked tables as well.
+     */
+    if (tables->prelocking_placeholder == TABLE_LIST::PRELOCK_FK)
+      continue;
+
+    auto *table= tables->table;
+    bool fks_prelocked= thd->locked_tables_mode <= LTM_LOCK_TABLES &&
+                        !has_prelocking_list &&
+                        table_can_need_prelocking(thd, tables);
+    /* Some tables will miss FK data in cases, like:
+       * Temporary tables
+       * Views
+       * Tables opened for read (SHOW CREATE will use TABLE_SHARE text data)
+       * Prelocking mode
+    */
+    if (!table)
+      continue;
+    table->foreign_keys= table->s->foreign_keys.elements;
+    table->foreign= NULL;
+    table->referenced_keys= table->s->referenced_keys.elements;
+    table->referenced= NULL;
+    if (!fks_prelocked)
+      continue;
+
+    table->foreign= convert_foreign_key_list(thd, table, NULL,
+                                             table->s->foreign_keys,
+                                             &table->mem_root);
+    table->referenced= convert_foreign_key_list(thd, NULL, table,
+                                                table->s->referenced_keys,
+                                                &table->mem_root);
+    
+  }
+
 #ifdef WITH_WSREP
   if (WSREP_ON                                         &&
       wsrep_replicate_myisam                           &&
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index d2ba46b4910..fca7f9713b1 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -195,6 +195,7 @@ Foreign_key::Foreign_key(const Foreign_key &rhs, MEM_ROOT *mem_root)
   ref_table(rhs.ref_table),
   ref_table_list(rhs.ref_table_list),
   ref_columns(rhs.ref_columns,mem_root),
+  ref_period(rhs.ref_period),
   delete_opt(rhs.delete_opt),
   update_opt(rhs.update_opt),
   match_opt(rhs.match_opt)
diff --git a/sql/sql_class.h b/sql/sql_class.h
index d0806f04cbc..3ba89db0ff0 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -372,11 +372,12 @@ public:
   Key(enum Keytype type_par, const LEX_CSTRING *name_arg,
       KEY_CREATE_INFO *key_info_arg,
       bool generated_arg, List<Key_part_spec> *cols,
+      Lex_ident period, bool without_overlaps,
       engine_option_value *create_opt, DDL_options_st ddl_options)
     :DDL_options(ddl_options),
      type(type_par), key_create_info(*key_info_arg), columns(*cols),
     name(*name_arg), option_list(create_opt), generated(generated_arg),
-    invisible(false), without_overlaps(false)
+    invisible(false), without_overlaps(without_overlaps), period(period)
   {}
   Key(const Key &rhs, MEM_ROOT *mem_root);
   virtual ~Key() {}
@@ -400,22 +401,23 @@ public:
   LEX_CSTRING ref_table;
   TABLE_LIST *ref_table_list;
   List<Key_part_spec> ref_columns;
+  Lex_ident ref_period;
   enum enum_fk_option delete_opt, update_opt;
   enum fk_match_opt match_opt;
   Foreign_key(const LEX_CSTRING *name_arg, List<Key_part_spec> *cols,
-              const LEX_CSTRING *constraint_name_arg,
+              Lex_ident period, const LEX_CSTRING *constraint_name_arg,
 	      const LEX_CSTRING *ref_db_arg, const LEX_CSTRING *ref_table_arg,
-	      TABLE_LIST *ref_table_list, List<Key_part_spec> *ref_cols,
+	      TABLE_LIST *ref_table_list, List<Key_part_spec> *ref_cols, Lex_ident ref_period,
 	       enum_fk_option delete_opt_arg,
 	      enum_fk_option update_opt_arg, fk_match_opt match_opt_arg,
 	      DDL_options ddl_options)
-    :Key(FOREIGN_KEY, name_arg, &default_key_create_info, 0, cols, NULL,
-         ddl_options),
+    :Key(FOREIGN_KEY, name_arg, &default_key_create_info, 0,
+         cols, period, false, NULL, ddl_options),
     constraint_name(*constraint_name_arg),
     ref_db(*ref_db_arg), ref_table(*ref_table_arg),
     ref_table_list(ref_table_list), ref_columns(*ref_cols),
-    delete_opt(delete_opt_arg), update_opt(update_opt_arg),
-    match_opt(match_opt_arg)
+    ref_period(ref_period), delete_opt(delete_opt_arg),
+    update_opt(update_opt_arg), match_opt(match_opt_arg)
    {
     // We don't check for duplicate FKs.
     key_create_info.check_for_duplicate_indexes= false;
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index 9724b60a336..3e86b655bbc 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -1289,6 +1289,7 @@ void LEX::start(THD *thd_arg)
 
   vers_conditions.empty();
   period_conditions.empty();
+  fk_ref_period= {};
 
   is_lex_started= TRUE;
 
@@ -11202,11 +11203,13 @@ bool LEX::add_table_foreign_key(const LEX_CSTRING *name,
 
   Key *key= new (thd->mem_root) Foreign_key(name,
                                             &last_key->columns,
+                                            last_key->period,
                                             constraint_name,
                                             &ref_table_name->db,
                                             &ref_table_name->table,
                                             ref_table,
                                             &ref_list,
+                                            fk_ref_period,
                                             fk_delete_opt,
                                             fk_update_opt,
                                             fk_match_option,
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 1c5b3bab558..d1038631e1d 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -3337,6 +3337,7 @@ public:
   enum Foreign_key::fk_match_opt fk_match_option;
   enum_fk_option fk_update_opt;
   enum_fk_option fk_delete_opt;
+  Lex_ident fk_ref_period;
   uint slave_thd_opt, start_transaction_opt;
   int nest_level;
   /*
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 70cf15e429e..ec6edf34eab 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -3707,6 +3707,38 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
                  ER_THD(thd, ER_KEY_REF_DO_NOT_MATCH_TABLE_REF));
 	DBUG_RETURN(TRUE);
       }
+
+      if (fk_key->ref_period)
+      {
+        auto *ref_table= fk_key->ref_table_list->table->s;
+        if (!fk_key->ref_period.streq(ref_table->period.name))
+        {
+          my_error(ER_PERIOD_FK_NOT_FOUND, MYF(0), fk_key->ref_period.str,
+                   ref_table->db.str, ref_table->table_name.str);
+        }
+
+        Create_field *period_start= NULL;
+        List_iterator_fast<Create_field> fit(alter_info->create_list);
+        while(auto *f= fit++)
+        {
+          if (create_info->period_info.period.start.streq(f->field_name))
+          {
+            period_start= f;
+            break;
+          }
+        }
+        DBUG_ASSERT(period_start);
+
+        auto *ref_period_start= ref_table->period.start_field(ref_table);
+
+        if (ref_period_start->type_handler() != period_start->type_handler()
+            || ref_period_start->pack_length() != period_start->pack_length)
+        {
+          my_error(ER_PERIOD_FK_TYPES_MISMATCH, MYF(0), fk_key->period.str,
+                   ref_table->db.str, ref_table->table_name.str,
+                   ref_table->period.name.str);
+        }
+      }
       continue;
     }
     (*key_count)++;
@@ -3956,6 +3988,30 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
         DBUG_RETURN(TRUE);
       }
 
+      switch (key->type) {
+        case Key::UNIQUE:
+        case Key::MULTIPLE:
+          if (!key->period)
+            break;
+          /* Fall through:
+             WITHOUT OVERLAPS and FOREIGN KEY with period forces fields to be
+             NOT NULL
+           */
+        case Key::PRIMARY:
+          /* Implicitly set primary key fields to NOT NULL for ISO conf. */
+          if (!(sql_field->flags & NOT_NULL_FLAG))
+          {
+            /* Implicitly set primary key fields to NOT NULL for ISO conf. */
+            sql_field->flags|= NOT_NULL_FLAG;
+            sql_field->pack_flag&= ~FIELDFLAG_MAYBE_NULL;
+            null_fields--;
+          }
+          break;
+        default:
+          // Fall through
+          break;
+      }
+
       cols2.rewind();
       switch(key->type) {
 
@@ -3993,13 +4049,6 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
 	                                                          *sql_field,
 	                                                          file))
           DBUG_RETURN(TRUE);
-        if (!(sql_field->flags & NOT_NULL_FLAG))
-        {
-          /* Implicitly set primary key fields to NOT NULL for ISO conf. */
-          sql_field->flags|= NOT_NULL_FLAG;
-          sql_field->pack_flag&= ~FIELDFLAG_MAYBE_NULL;
-          null_fields--;
-        }
         break;
 
       case Key::MULTIPLE:
@@ -4549,7 +4598,7 @@ static bool append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
   Key *key= NULL;
   while ((key=key_it++))
   {
-    if (key->type != Key::PRIMARY && key->type != Key::UNIQUE)
+    if (!key->period && key->type != Key::PRIMARY && key->type != Key::UNIQUE)
       continue;
 
     if (create_info->versioned())
@@ -4562,8 +4611,19 @@ static bool append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
             row_end_field.streq(key_part->field_name))
           break;
       }
-      if (!key_part)
+      if (!key_part || key->type == Key::FOREIGN_KEY)
         key->columns.push_back(new Key_part_spec(&row_end_field, 0));
+
+      if (key->type == Key::FOREIGN_KEY)
+      {
+        auto *fk = static_cast<Foreign_key *>(key);
+
+        const LEX_CSTRING *ref_vers_end= &row_end_field;
+        if (fk->ref_table_list->table)
+          ref_vers_end= &fk->ref_table_list->table->vers_end_field()->field_name;
+
+        fk->ref_columns.push_back(new Key_part_spec(ref_vers_end, 0));
+      }
     }
   }
 
@@ -4603,6 +4663,30 @@ static bool append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
       key->columns.push_back(new Key_part_spec(&period_start, 0));
       key->columns.push_back(new Key_part_spec(&period_end, 0));
     }
+    else if (key->period)
+    {
+      if (!create_info->period_info.is_set()
+          || !key->period.streq(create_info->period_info.name))
+      {
+        my_error(ER_PERIOD_NOT_FOUND, MYF(0), key->period.str);
+        return true;
+      }
+      const auto &period_start= create_info->period_info.period.start;
+      const auto &period_end= create_info->period_info.period.end;
+      key->columns.push_back(new Key_part_spec(&period_start, 0));
+      key->columns.push_back(new Key_part_spec(&period_end, 0));
+
+      if (key->type == Key::FOREIGN_KEY)
+      {
+        auto *fk= static_cast<Foreign_key*>(key);
+        const auto &ref_period= fk->ref_table_list->table->s->period;
+        const auto *field= fk->ref_table_list->table->field;
+        const auto &ref_period_start= field[ref_period.start_fieldno]->field_name;
+        const auto &ref_period_end= field[ref_period.end_fieldno]->field_name;
+        fk->ref_columns.push_back(new Key_part_spec(&ref_period_start, 0));
+        fk->ref_columns.push_back(new Key_part_spec(&ref_period_end, 0));
+      }
+    }
   }
 
   return false;
@@ -8616,9 +8700,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       /* We dont need LONG_UNIQUE_HASH_FIELD flag because it will be autogenerated */
       key= new (thd->mem_root) Key(key_type, &tmp_name, &key_create_info,
                    MY_TEST(key_info->flags & HA_GENERATED_KEY),
-                   &key_parts, key_info->option_list, DDL_options());
-      key->without_overlaps= key_info->without_overlaps;
-      key->period= table->s->period.name;
+                   &key_parts, table->s->period.name, key_info->without_overlaps,
+                   key_info->option_list, DDL_options());
       new_key_list.push_back(key, thd->mem_root);
     }
     if (long_hash_key)
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index 323f693302f..7a770f8a0c5 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -1450,15 +1450,14 @@ Type_handler_hybrid_field_type::Type_handler_hybrid_field_type()
 /***************************************************************************/
 
 /* number of bytes to store second_part part of the TIMESTAMP(N) */
-uint Type_handler_timestamp::m_sec_part_bytes[MAX_DATETIME_PRECISION + 1]=
+const uint Type_handler_timestamp::m_sec_part_bytes[MAX_DATETIME_PRECISION + 1]=
      { 0, 1, 1, 2, 2, 3, 3 };
 
 /* number of bytes to store DATETIME(N) */
-uint Type_handler_datetime::m_hires_bytes[MAX_DATETIME_PRECISION + 1]=
-     { 5, 6, 6, 7, 7, 7, 8 };
+constexpr uint Type_handler_datetime::m_hires_bytes[MAX_DATETIME_PRECISION + 1];
 
 /* number of bytes to store TIME(N) */
-uint Type_handler_time::m_hires_bytes[MAX_DATETIME_PRECISION + 1]=
+const uint Type_handler_time::m_hires_bytes[MAX_DATETIME_PRECISION + 1]=
      { 3, 4, 4, 5, 5, 5, 6 };
 
 /***************************************************************************/
diff --git a/sql/sql_type.h b/sql/sql_type.h
index 3943a3f761f..0ae46b9560a 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -5806,7 +5806,7 @@ public:
 class Type_handler_time: public Type_handler_time_common
 {
   /* number of bytes to store TIME(N) */
-  static uint m_hires_bytes[MAX_DATETIME_PRECISION+1];
+  static const uint m_hires_bytes[MAX_DATETIME_PRECISION+1];
 public:
   static uint hires_bytes(uint dec) { return m_hires_bytes[dec]; }
   virtual ~Type_handler_time() {}
@@ -6105,9 +6105,10 @@ public:
 class Type_handler_datetime: public Type_handler_datetime_common
 {
   /* number of bytes to store DATETIME(N) */
-  static uint m_hires_bytes[MAX_DATETIME_PRECISION + 1];
+  static constexpr uint m_hires_bytes[MAX_DATETIME_PRECISION + 1]=
+          { 5, 6, 6, 7, 7, 7, 8 };
 public:
-  static uint hires_bytes(uint dec) { return m_hires_bytes[dec]; }
+  static constexpr uint hires_bytes(uint dec) { return m_hires_bytes[dec]; }
   virtual ~Type_handler_datetime() {}
   const Name version() const override { return version_mariadb53(); }
   uint32 max_display_length_for_field(const Conv_source &src) const override
@@ -6267,7 +6268,7 @@ public:
 class Type_handler_timestamp: public Type_handler_timestamp_common
 {
   /* number of bytes to store second_part part of the TIMESTAMP(N) */
-  static uint m_sec_part_bytes[MAX_DATETIME_PRECISION + 1];
+  static const uint m_sec_part_bytes[MAX_DATETIME_PRECISION + 1];
 public:
   static uint sec_part_bytes(uint dec) { return m_sec_part_bytes[dec]; }
   virtual ~Type_handler_timestamp() {}
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 7a4367b96c7..7511eda6357 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -5826,7 +5826,7 @@ key_def:
               MYSQL_YYABORT;
             Lex->option_list= NULL;
           }
-          '(' key_list ')' references
+          '(' key_list_fk ')' references
           {
             if (unlikely(Lex->add_table_foreign_key($5.str ? &$5 : &$1,
                                                     $1.str ? &$1 : &$5, $10, $4)))
@@ -6777,6 +6777,10 @@ ref_list:
               MYSQL_YYABORT;
             Lex->ref_list.push_back(key, thd->mem_root);
           }
+        | ref_list ',' PERIOD_SYM ident
+          {
+            Lex->fk_ref_period= $4;
+          }
         | ident
           {
             Key_part_spec *key= new (thd->mem_root) Key_part_spec(&$1, 0);
@@ -6997,6 +7001,21 @@ btree_or_rtree:
         | HASH_SYM  { $$= HA_KEY_ALG_HASH; }
         ;
 
+key_list_fk:
+          key_list_fk ',' key_part order_dir
+          {
+            Lex->last_key->columns.push_back($3, thd->mem_root);
+          }
+        | key_list_fk ',' PERIOD_SYM ident
+          {
+            Lex->last_key->period= $4;
+          }
+        | key_part order_dir
+          {
+            Lex->last_key->columns.push_back($1, thd->mem_root);
+          }
+        ;
+
 key_list:
           key_list ',' key_part order_dir
           {
diff --git a/sql/table.cc b/sql/table.cc
index 33480ba4bd7..6d50c94087d 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -1773,7 +1773,7 @@ int TABLE_SHARE::init_from_binary_frm_image(THD *thd, bool write,
 
   if (frm_length < FRM_HEADER_SIZE + len ||
       !(pos= uint4korr(frm_image + FRM_HEADER_SIZE + len)))
-    goto err;
+    DBUG_RETURN(err());
 
   if (read_extra2(frm_image, len, &extra2))
     DBUG_RETURN(err());
@@ -9388,6 +9388,7 @@ bool TABLE_SHARE::update_foreign_keys(THD *thd, Alter_info *alter_info,
     dst->delete_method= src->delete_opt;
     dst->foreign_fields.empty();
     dst->referenced_fields.empty();
+    dst->has_period= bool(src->period);
 
     Key_part_spec* col;
     List_iterator_fast<Key_part_spec> col_it(src->columns);
diff --git a/sql/table.h b/sql/table.h
index 01c06e10405..70e5f371546 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1136,6 +1136,7 @@ struct st_cond_statistic;
 class SplM_opt_info;
 
 struct vers_select_conds_t;
+struct FOREIGN_KEY;
 
 struct TABLE
 {
@@ -1201,6 +1202,11 @@ public:
   Field *found_next_number_field;	/* Set on open */
   Virtual_column_info **check_constraints;
 
+  uint foreign_keys;
+  uint referenced_keys;
+  FOREIGN_KEY *foreign;
+  FOREIGN_KEY *referenced;
+
   /* Table's triggers, 0 if there are no of them */
   Table_triggers_list *triggers;
   TABLE_LIST *pos_in_table_list;/* Element referring to this table */
@@ -1719,11 +1725,24 @@ typedef struct st_foreign_key_info
   LEX_CSTRING *referenced_key_name;
   List<LEX_CSTRING> foreign_fields;
   List<LEX_CSTRING> referenced_fields;
+  bool has_period;
 } FOREIGN_KEY_INFO;
 
 LEX_CSTRING *fk_option_name(enum_fk_option opt);
 bool fk_modifies_child(enum_fk_option opt);
 
+struct FOREIGN_KEY
+{
+  uint foreign_key_nr;
+  uint referenced_key_nr;
+  KEY *foreign_key;
+  KEY *referenced_key;
+  uint fields_num;
+  bool has_period;
+  enum_fk_option update_method;
+  enum_fk_option delete_method;
+};
+
 class IS_table_read_plan;
 
 /*
diff --git a/storage/innobase/dict/dict0crea.cc b/storage/innobase/dict/dict0crea.cc
index 19bf3b33cfa..5186588f253 100644
--- a/storage/innobase/dict/dict0crea.cc
+++ b/storage/innobase/dict/dict0crea.cc
@@ -1762,6 +1762,8 @@ dict_foreign_def_get(
 		}
 	}
 
+	// TODO get period name
+
 	strcat(fk_def,(char *)") REFERENCES ");
 
 	bufend = innobase_convert_name(tablebuf, MAX_TABLE_NAME_LEN,
@@ -1847,14 +1849,14 @@ dict_create_add_foreign_to_dictionary(
 	pars_info_add_str_literal(info, "ref_name",
 				  foreign->referenced_table_name);
 
-	pars_info_add_int4_literal(info, "n_cols",
-				   ulint(foreign->n_fields)
-				   | (ulint(foreign->type) << 24));
+	uint32_t n_cols = ulint(foreign->n_fields)
+			  | (ulint(foreign->type) << 24U)
+			  | (ulint(foreign->has_period) << 30U);
+	pars_info_add_int4_literal(info, "n_cols", n_cols);
 
 	DBUG_PRINT("dict_create_add_foreign_to_dictionary",
 		   ("'%s', '%s', '%s', %d", foreign->id, name,
-		    foreign->referenced_table_name,
-		    foreign->n_fields + (foreign->type << 24)));
+		    foreign->referenced_table_name, n_cols));
 
 	error = dict_foreign_eval_sql(info,
 				      "PROCEDURE P () IS\n"
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 1d0baec2915..20c70686eb9 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -2811,6 +2811,10 @@ dict_foreign_find_index(
 					/*!< in: nonzero if none of
 					the columns must be declared
 					NOT NULL */
+	bool			check_period,
+					/*!< in: check if index contains
+					an application-time period
+					without overlaps*/
 	fkerr_t*		error,	/*!< out: error code */
 	ulint*			err_col_no,
 					/*!< out: column number where
@@ -2834,7 +2838,7 @@ dict_foreign_find_index(
 		    && dict_foreign_qualify_index(
 			    table, col_names, columns, n_cols,
 			    index, types_idx,
-			    check_charsets, check_null,
+			    check_charsets, check_null, check_period,
 			    error, err_col_no, err_index)) {
 			if (error) {
 				*error = FK_SUCCESS;
@@ -2946,7 +2950,7 @@ dict_foreign_add_to_cache(
 			ref_table, NULL,
 			for_in_cache->referenced_col_names,
 			for_in_cache->n_fields, for_in_cache->foreign_index,
-			check_charsets, false);
+			check_charsets, false, false);
 
 		if (index == NULL
 		    && !(ignore_err & DICT_ERR_IGNORE_FK_NOKEY)) {
@@ -2985,7 +2989,8 @@ dict_foreign_add_to_cache(
 			for_in_cache->referenced_index, check_charsets,
 			for_in_cache->type
 			& (DICT_FOREIGN_ON_DELETE_SET_NULL
-			   | DICT_FOREIGN_ON_UPDATE_SET_NULL));
+			   | DICT_FOREIGN_ON_UPDATE_SET_NULL),
+                        false);
 
 		if (index == NULL
 		    && !(ignore_err & DICT_ERR_IGNORE_FK_NOKEY)) {
@@ -4436,7 +4441,7 @@ dict_foreign_replace_index(
 				foreign->foreign_col_names,
 				foreign->n_fields, index,
 				/*check_charsets=*/TRUE, /*check_null=*/FALSE,
-				NULL, NULL, NULL);
+				false, NULL, NULL, NULL);
 			if (new_index) {
 				ut_ad(new_index->table == index->table);
 				ut_ad(!new_index->to_be_dropped);
@@ -4461,7 +4466,7 @@ dict_foreign_replace_index(
 				foreign->referenced_col_names,
 				foreign->n_fields, index,
 				/*check_charsets=*/TRUE, /*check_null=*/FALSE,
-				NULL, NULL, NULL);
+				false, NULL, NULL, NULL);
 			/* There must exist an alternative index,
 			since this must have been checked earlier. */
 			if (new_index) {
@@ -4965,6 +4970,10 @@ dict_foreign_qualify_index(
 					/*!< in: nonzero if none of
 					the columns must be declared
 					NOT NULL */
+	bool			check_period,
+					/*!< in: check if index contains
+					an application-time period
+					without overlaps*/
 	fkerr_t*		error,	/*!< out: error code */
 	ulint*			err_col_no,
 					/*!< out: column number where
@@ -4977,6 +4986,26 @@ dict_foreign_qualify_index(
 		return(false);
 	}
 
+	if (check_period) {
+		if ((index->type & DICT_PERIOD) == 0) {
+			return(false);
+		}
+
+		/* Despite it is theoretically possible to construct such
+		   an index with period not at the last positions,
+		   it is not supported at least for now.
+		*/
+		if (dict_index_get_n_ordering_defined_by_user(index) != n_cols){
+			return(false);
+		}
+		auto pstart = dict_index_get_nth_field(index, n_cols - 2);
+		auto pend = dict_index_get_nth_field(index, n_cols - 1);
+		if ((pstart->col->prtype & DATA_PERIOD_START) == 0
+		    || (pend->col->prtype & DATA_PERIOD_END) == 0) {
+			return false;
+		}
+	}
+
 	if (index->type & (DICT_SPATIAL | DICT_FTS)) {
 		return false;
 	}
diff --git a/storage/innobase/dict/dict0load.cc b/storage/innobase/dict/dict0load.cc
index ebb8c4bc24f..7bd25e11b9a 100644
--- a/storage/innobase/dict/dict0load.cc
+++ b/storage/innobase/dict/dict0load.cc
@@ -2503,7 +2503,8 @@ dict_load_indexes(
 		subsequent checks are relevant for the supported types. */
 		if (index->type & ~(DICT_CLUSTERED | DICT_UNIQUE
 				    | DICT_CORRUPT | DICT_FTS
-				    | DICT_SPATIAL | DICT_VIRTUAL)) {
+				    | DICT_SPATIAL | DICT_VIRTUAL
+				    | DICT_PERIOD)) {
 
 			ib::error() << "Unknown type " << index->type
 				<< " of index " << index->name
@@ -3455,8 +3456,9 @@ dict_load_foreign(
 
 	/* We store the type in the bits 24..29 of n_fields_and_type. */
 
-	foreign->type = (unsigned int) (n_fields_and_type >> 24);
+	foreign->type = (unsigned int) (n_fields_and_type >> 24U);
 	foreign->n_fields = (unsigned int) (n_fields_and_type & 0x3FFUL);
+	foreign->has_period = (unsigned) (n_fields_and_type >> 30U);
 
 	foreign->id = mem_heap_strdupl(foreign->heap, id, id_len);
 
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index 811c9a0dc03..0176741958c 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -342,6 +342,13 @@ dict_mem_table_add_col(
 		ut_ad(!table->vers_end);
 		table->vers_end = i;
 	}
+	if (prtype & DATA_PERIOD_START) {
+		table->period_start = i;
+		table->has_period = true;
+	} else if (prtype & DATA_PERIOD_END) {
+		table->period_end = i;
+		table->has_period = true;
+	}
 }
 
 /** Adds a virtual column definition to a table.
@@ -603,7 +610,7 @@ dict_mem_table_col_rename_low(
 			dict_index_t* new_index = dict_foreign_find_index(
 				foreign->foreign_table, NULL,
 				foreign->foreign_col_names,
-				foreign->n_fields, NULL, true, false,
+				foreign->n_fields, NULL, true, false, false,
 				NULL, NULL, NULL);
 			/* There must be an equivalent index in this case. */
 			ut_ad(new_index != NULL);
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 6c3d5e3fb38..4ee84a1a3b0 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -10037,7 +10037,7 @@ wsrep_append_foreign_key(
 						foreign->referenced_col_names,
 						foreign->n_fields,
 						foreign->foreign_index,
-						TRUE, FALSE);
+						true, false, false);
 			}
 		} else {
 	  		foreign->foreign_table =
@@ -10051,7 +10051,7 @@ wsrep_append_foreign_key(
 						foreign->foreign_col_names,
 						foreign->n_fields,
 						foreign->referenced_index,
-						TRUE, FALSE);
+						true, false, false);
 			}
 		}
 		mutex_exit(&dict_sys.mutex);
@@ -10736,16 +10736,23 @@ create_table_info_t::create_table_def()
 
 	for (ulint i = 0, j = 0; j < n_cols; i++) {
 		Field*	field = m_form->field[i];
-		ulint vers_row = 0;
+		ulint vers_period_row = 0;
 
 		if (m_form->versioned()) {
 			if (i == m_form->s->vers.start_fieldno) {
-				vers_row = DATA_VERS_START;
+				vers_period_row = DATA_VERS_START;
 			} else if (i == m_form->s->vers.end_fieldno) {
-				vers_row = DATA_VERS_END;
+				vers_period_row = DATA_VERS_END;
 			} else if (!(field->flags
 				     & VERS_UPDATE_UNVERSIONED_FLAG)) {
-				vers_row = DATA_VERSIONED;
+				vers_period_row = DATA_VERSIONED;
+			}
+		}
+		if (m_form->s->period.name) {
+			if (i == m_form->s->period.start_fieldno) {
+				vers_period_row |= DATA_PERIOD_START;
+			} else if (i == m_form->s->period.end_fieldno) {
+				vers_period_row |= DATA_PERIOD_END;
 			}
 		}
 
@@ -10833,7 +10840,7 @@ err_col:
 					(ulint) field->type()
 					| nulls_allowed | unsigned_type
 					| binary_type | long_true_varchar
-					| vers_row,
+					| vers_period_row,
 					charset_no),
 				col_len);
 		} else if (!omit_virtual) {
@@ -10843,7 +10850,7 @@ err_col:
 					(ulint) field->type()
 					| nulls_allowed | unsigned_type
 					| binary_type | long_true_varchar
-					| vers_row
+					| vers_period_row
 					| is_virtual,
 					charset_no),
 				col_len, i, 0);
@@ -11001,6 +11008,7 @@ create_index(
 		/* Only one of these can be specified at a time. */
 		ut_ad(~key->flags & (HA_SPATIAL | HA_FULLTEXT));
 		ut_ad(!(key->flags & HA_NOSAME));
+		ut_ad(!key->without_overlaps);
 		index = dict_mem_index_create(table, key->name.str,
 					      (key->flags & HA_SPATIAL)
 					      ? DICT_SPATIAL : DICT_FTS,
@@ -11036,6 +11044,11 @@ create_index(
 		ind_type |= DICT_UNIQUE;
 	}
 
+	if (key->without_overlaps) {
+		ut_ad(ind_type & DICT_UNIQUE);
+		ind_type |= DICT_PERIOD;
+	}
+
 	field_lengths = (ulint*) my_malloc(//PSI_INSTRUMENT_ME,
 		key->user_defined_key_parts * sizeof *
 				field_lengths, MYF(MY_FAE));
@@ -12290,6 +12303,41 @@ public:
 	const char* str() { return buf; }
 };
 
+static const unsigned MAX_COLS_PER_FK = 500;
+
+static const char *get_fk_column(dict_table_t *table, trx_t *trx,
+				 dict_foreign_t *foreign, FOREIGN_KEY_INFO *fk,
+				 const LEX_CSTRING &field_name,
+				 char *create_name,
+				 const char *operation,
+				 int ncol)
+{
+	const char *column_name= mem_heap_strdupl(foreign->heap,
+						  field_name.str,
+						  field_name.length);
+	bool success = find_col(table, &column_name);
+	if (!success) {
+		key_text k(fk);
+		ib_foreign_warn(trx, DB_CANNOT_ADD_CONSTRAINT, create_name,
+				"%s table %s foreign key %s constraint failed."
+				" Column %s was not found.",
+				operation, create_name, k.str(),
+				column_name);
+
+		return NULL;
+	}
+	if (ncol >= MAX_COLS_PER_FK - 1) {
+		key_text k(fk);
+		ib_foreign_warn(trx, DB_CANNOT_ADD_CONSTRAINT, create_name,
+				"%s table %s foreign key %s constraint failed."
+				" Too many columns: %u (%u allowed).",
+				operation, create_name, k.str(), ncol,
+				MAX_COLS_PER_FK);
+		return NULL;
+	}
+	return column_name;
+}
+
 /** Create InnoDB foreign keys from MySQL alter_info. Collect all
 dict_foreign_t items into local_fk_set and then add into system table.
 @return		DB_SUCCESS or specific error code */
@@ -12300,7 +12348,6 @@ create_table_info_t::create_foreign_keys()
 	dict_foreign_set_free local_fk_set_free(local_fk_set);
 	dberr_t		      error;
 	ulint		      number	      = 1;
-	static const unsigned MAX_COLS_PER_FK = 500;
 	const char*	      column_names[MAX_COLS_PER_FK];
 	const char*	      ref_column_names[MAX_COLS_PER_FK];
 	char		      create_name[MAX_TABLE_NAME_LEN + 1];
@@ -12382,7 +12429,7 @@ create_table_info_t::create_foreign_keys()
 		}
 
 		LEX_CSTRING*   col;
-		bool	       success;
+		bool	       success = true;
 
 		dict_foreign_t* foreign = dict_mem_foreign_create();
 		if (!foreign) {
@@ -12431,38 +12478,16 @@ create_table_info_t::create_foreign_keys()
 		List_iterator_fast<LEX_CSTRING> col_it(fk->foreign_fields);
 		unsigned			  i = 0, j = 0;
 		while ((col = col_it++)) {
-			column_names[i] = mem_heap_strdupl(
-				foreign->heap, col->str, col->length);
-			success = find_col(table, column_names + i);
-			if (!success) {
-				key_text k(fk);
-				ib_foreign_warn(
-					m_trx, DB_CANNOT_ADD_CONSTRAINT,
-					create_name,
-					"%s table %s foreign key %s constraint"
-					" failed. Column %s was not found.",
-					operation, create_name, k.str(),
-					column_names[i]);
-
-				return (DB_CANNOT_ADD_CONSTRAINT);
-			}
-			++i;
-			if (i >= MAX_COLS_PER_FK) {
-				key_text k(fk);
-				ib_foreign_warn(
-					m_trx, DB_CANNOT_ADD_CONSTRAINT,
-					create_name,
-					"%s table %s foreign key %s constraint"
-					" failed. Too many columns: %u (%u "
-					"allowed).",
-					operation, create_name, k.str(), i,
-					MAX_COLS_PER_FK);
-				return (DB_CANNOT_ADD_CONSTRAINT);
-			}
+			column_names[i] = get_fk_column(table, m_trx, foreign,
+							fk, *col, create_name,
+							operation, i);
+			if (column_names[i] == nullptr)
+				return DB_CANNOT_ADD_CONSTRAINT;
+			i++;
 		}
 
 		index = dict_foreign_find_index(
-			table, NULL, column_names, i, NULL, TRUE, FALSE,
+			table, NULL, column_names, i, NULL, true, FALSE, false,
 			&index_error, &err_col, &err_index);
 
 		if (!index) {
@@ -12485,6 +12510,7 @@ create_table_info_t::create_foreign_keys()
 
 		foreign->foreign_index = index;
 		foreign->n_fields      = (unsigned int)i;
+		foreign->has_period    = fk->has_period;
 
 		foreign->foreign_col_names = static_cast<const char**>(
 			mem_heap_alloc(foreign->heap, i * sizeof(void*)));
@@ -12540,23 +12566,25 @@ create_table_info_t::create_foreign_keys()
 			if (foreign->referenced_table) {
 				success = find_col(foreign->referenced_table,
 						   ref_column_names + j);
-				if (!success) {
-					key_text k(fk);
-					ib_foreign_warn(
-						m_trx,
-						DB_CANNOT_ADD_CONSTRAINT,
-						create_name,
-						"%s table %s foreign key %s "
-						"constraint failed. "
-						"Column %s was not found.",
-						operation, create_name,
-						k.str(), ref_column_names[j]);
-
-					return (DB_CANNOT_ADD_CONSTRAINT);
-				}
+				if (!success)
+					break;
 			}
 			++j;
 		}
+
+		if (!success)
+		{
+			key_text k(fk);
+			ib_foreign_warn(m_trx,
+					DB_CANNOT_ADD_CONSTRAINT, create_name,
+					"%s table %s foreign key %s "
+					"constraint failed. "
+					"Column %s was not found.",
+					operation, create_name,
+					k.str(), ref_column_names[j]);
+
+			return DB_CANNOT_ADD_CONSTRAINT;
+		}
 		/* See ER_WRONG_FK_DEF in mysql_prepare_create_table() */
 		ut_ad(i == j);
 
@@ -12568,8 +12596,8 @@ create_table_info_t::create_foreign_keys()
 			index = dict_foreign_find_index(
 				foreign->referenced_table, NULL,
 				ref_column_names, i, foreign->foreign_index,
-				TRUE, FALSE, &index_error, &err_col,
-				&err_index);
+				TRUE, FALSE, fk->has_period,
+				&index_error, &err_col, &err_index);
 
 			if (!index) {
 				key_text k(fk);
@@ -15509,6 +15537,9 @@ get_foreign_key_info(
 	}
 
 	f_key_info.referenced_key_name = referenced_key_name;
+	f_key_info.has_period = foreign->has_period;
+	bool has_period = f_key_info.has_period;
+
 	return pf_key_info;
 }
 
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index db1b8257724..38e1eea50c9 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -496,6 +496,7 @@ inline bool dict_table_t::instant_column(const dict_table_t& table,
 			c.def_val = o->def_val;
 			DBUG_ASSERT(!((c.prtype ^ o->prtype)
 				      & ~(DATA_NOT_NULL | DATA_VERSIONED
+				          | DATA_PERIOD_START | DATA_PERIOD_END
 					  | CHAR_COLL_MASK << 16
 					  | DATA_LONG_TRUE_VARCHAR)));
 			DBUG_ASSERT(c.same_type(*o));
@@ -2710,7 +2711,7 @@ innobase_find_fk_index(
 	while (index != NULL) {
 		if (dict_foreign_qualify_index(table, col_names, columns,
 					       n_cols, index, NULL, true, 0,
-					       NULL, NULL, NULL)
+					       false, NULL, NULL, NULL)
 		    && std::find(drop_index.begin(), drop_index.end(), index)
 			   == drop_index.end()) {
 			return index;
@@ -2919,7 +2920,7 @@ innobase_get_foreign_key_info(
 						referenced_table, 0,
 						referenced_column_names,
 						i, index,
-						TRUE, FALSE,
+						TRUE, FALSE, false,
 						NULL, NULL, NULL);
 
 				DBUG_EXECUTE_IF(
@@ -7071,7 +7072,7 @@ innobase_check_foreign_key_index(
 			    foreign->n_fields, index,
 			    /*check_charsets=*/TRUE,
 			    /*check_null=*/FALSE,
-			    NULL, NULL, NULL)
+			    false, NULL, NULL, NULL)
 		    && NULL == innobase_find_equiv_index(
 			    foreign->referenced_col_names,
 			    foreign->n_fields,
@@ -7106,7 +7107,7 @@ innobase_check_foreign_key_index(
 			    foreign->n_fields, index,
 			    /*check_charsets=*/TRUE,
 			    /*check_null=*/FALSE,
-			    NULL, NULL, NULL)
+			    false, NULL, NULL, NULL)
 		    && NULL == innobase_find_equiv_index(
 			    foreign->foreign_col_names,
 			    foreign->n_fields,
@@ -9381,7 +9382,7 @@ innobase_update_foreign_try(
 				fk->type
 				& (DICT_FOREIGN_ON_DELETE_SET_NULL
 					| DICT_FOREIGN_ON_UPDATE_SET_NULL),
-				NULL, NULL, NULL);
+				false, NULL, NULL, NULL);
 			if (!fk->foreign_index) {
 				my_error(ER_FK_INCORRECT_OPTION,
 					 MYF(0), table_name, fk->id);
diff --git a/storage/innobase/include/data0type.h b/storage/innobase/include/data0type.h
index 0e496085113..d783d097d6f 100644
--- a/storage/innobase/include/data0type.h
+++ b/storage/innobase/include/data0type.h
@@ -195,6 +195,12 @@ be less than 256 */
 /** system-versioned user data column */
 #define DATA_VERSIONED (DATA_VERS_START|DATA_VERS_END)
 
+/** Application-time periods */
+#define DATA_PERIOD_START	65536U	/* start system field */
+#define DATA_PERIOD_END		131072U	/* end system field */
+
+#define DATA_PRTYPE_MAX DATA_PERIOD_END
+
 /** Check whether locking is disabled (never). */
 #define dict_table_is_locking_disabled(table) false
 
@@ -353,7 +359,7 @@ UNIV_INLINE
 uint32_t
 dtype_form_prtype(ulint old_prtype, ulint charset_coll)
 {
-	ut_ad(old_prtype < 256 * 256);
+	ut_ad(old_prtype < DATA_PRTYPE_MAX << 1);
 	ut_ad(charset_coll <= MAX_CHAR_COLL_NUM);
 	return(uint32_t(old_prtype + (charset_coll << 16)));
 }
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index 70823dae7f4..f8eb13c6f33 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -542,6 +542,10 @@ dict_foreign_find_index(
 					/*!< in: nonzero if none of
 					the columns must be declared
 					NOT NULL */
+	bool			check_period,
+					/*!< in: check if index contains
+					an application-time period
+					without overlaps*/
 	fkerr_t*		error = NULL,	/*!< out: error code */
 	ulint*			err_col_no = NULL,
 					/*!< out: column number where
@@ -622,6 +626,10 @@ dict_foreign_qualify_index(
 					/*!< in: nonzero if none of
 					the columns must be declared
 					NOT NULL */
+	bool			check_period,
+					/*!< in: check if index contains
+					an application-time period
+					without overlaps*/
 	fkerr_t*		error,	/*!< out: error code */
 	ulint*			err_col_no,
 					/*!< out: column number where
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 4f633458fe6..a90972340f1 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -72,7 +72,10 @@ combination of types */
 				other flags */
 #define	DICT_VIRTUAL	128	/* Index on Virtual column */
 
-#define	DICT_IT_BITS	8	/*!< number of bits used for
+#define	DICT_PERIOD	256	/* Last two user fields treated as period
+				of the unique index */
+
+#define	DICT_IT_BITS	9	/*!< number of bits used for
 				SYS_INDEXES.TYPE */
 /* @} */
 
@@ -731,6 +734,7 @@ public:
 			&& mbmaxlen == other.mbmaxlen
 			&& !((prtype ^ other.prtype)
 			     & ~(DATA_NOT_NULL | DATA_VERSIONED
+			         | DATA_PERIOD_START | DATA_PERIOD_END
 				 | CHAR_COLL_MASK << 16
 				 | DATA_LONG_TRUE_VARCHAR));
 	}
@@ -1374,6 +1378,11 @@ struct dict_foreign_t{
 					as the first fields are as mentioned */
 	unsigned	type:6;		/*!< 0 or DICT_FOREIGN_ON_DELETE_CASCADE
 					or DICT_FOREIGN_ON_DELETE_SET_NULL */
+	bool		has_period:1;	/*!< true if a reference contains an
+ 					Application-time period.
+ 					The referenced key should be
+ 					WITHOUT OVERLAPS, thus, flagged with
+ 					DICT_PERIOD */
 	char*		foreign_table_name;/*!< foreign table name */
 	char*		foreign_table_name_lookup;
 				/*!< foreign table name used for dict lookup */
@@ -2018,6 +2027,13 @@ public:
 				/*!< System Versioning: row start col index */
 	unsigned	vers_end:10;
 				/*!< System Versioning: row end col index */
+	unsigned	period_start:10;
+				/*!< Period start column index */
+	unsigned	period_end:10;
+				/*!< Period end column index */
+	bool		has_period:1;
+				/*!< True if table has period. In this case,
+				 * period_start and period_end have a meaning */
 	bool		is_system_db;
 				/*!< True if the table belongs to a system
 				database (mysql, information_schema or
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 735bd4517a1..64df17d0e08 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -1595,6 +1595,10 @@ row_ins_check_foreign_constraint(
 		goto exit_func;
 	}
 
+	/* Temporal referential constraints are handled on sql layer */
+	if (foreign->has_period)
+		goto exit_func;
+
 	/* If any of the foreign key fields in entry is SQL NULL, we
 	suppress the foreign key check: this is compatible with Oracle,
 	for example */