[WIP] MDEV-16983 Application-time periods: foreign key PART 1/4nikita/periods-fk-rebase

30 files changed, 1439 insertions, 146 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..a7286865b8b
--- /dev/null
+++ b/mysql-test/suite/period/r/fk.result
@@ -0,0 +1,309 @@
+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) DEFAULT NULL,
+  `x` int(11) DEFAULT NULL,
+  `s` date NOT NULL,
+  `e` date NOT NULL,
+  PERIOD FOR `fp` (`s`, `e`),
+  KEY `id` (`id`,`x`,`e`,`s`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`id`, `x`, period `fp`) REFERENCES `t` (`id`, `x`, period `p`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+flush tables;
+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
+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);
+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) DEFAULT NULL,
+  `z` char(200) DEFAULT NULL,
+  `pk` int(11) DEFAULT 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`,`e`,`s`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`pk`, `y`, period `fp`) REFERENCES `t` (`id`, `x`, period `p`)
+) 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');
+insert into s values ( 0, 1, '1', 1, '2017-02-20', '2017-02-03');
+ERROR 23000: Cannot add or update a child row: a foreign key constraint fails
+delete from t;
+ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails
+delete from s;
+delete from t;
+call mtr.add_suppression("In ALTER TABLE .* has or is referenced in foreign key constraints which are not compatible with the new table definition.");
+alter table t drop period for p, drop constraint id;
+ERROR HY000: Error on rename of 'OLD_FILE_NAME' to 'NEW_FILE_NAME' (errno: 150 "Foreign key constraint is incorrectly formed")
+show create table s;
+Table	Create Table
+s	CREATE TABLE `s` (
+  `x` int(11) DEFAULT NULL,
+  `y` int(11) DEFAULT NULL,
+  `z` char(200) DEFAULT NULL,
+  `pk` int(11) DEFAULT 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`,`e`,`s`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`pk`, `y`, period `fp`) REFERENCES `t` (`id`, `x`, period `p`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+alter table t drop period for p, drop constraint id, add period for pop(s, e),
+add unique(id, x, pop without overlaps);
+show create table s;
+Table	Create Table
+s	CREATE TABLE `s` (
+  `x` int(11) DEFAULT NULL,
+  `y` int(11) DEFAULT NULL,
+  `z` char(200) DEFAULT NULL,
+  `pk` int(11) DEFAULT 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`,`e`,`s`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`pk`, `y`, period `fp`) REFERENCES `t` (`id`, `x`, period `pop`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+alter table t drop period for pop, drop constraint id,
+add unique(id, x, s, e);
+ERROR HY000: Error on rename of 'OLD_FILE_NAME' to 'NEW_FILE_NAME' (errno: 150 "Foreign key constraint is incorrectly formed")
+show create table s;
+Table	Create Table
+s	CREATE TABLE `s` (
+  `x` int(11) DEFAULT NULL,
+  `y` int(11) DEFAULT NULL,
+  `z` char(200) DEFAULT NULL,
+  `pk` int(11) DEFAULT 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`,`e`,`s`),
+  CONSTRAINT `s_ibfk_1` FOREIGN KEY (`pk`, `y`, period `fp`) REFERENCES `t` (`id`, `x`, period `pop`)
+) ENGINE=InnoDB DEFAULT CHARSET=latin1
+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..d388d29e162
--- /dev/null
+++ b/mysql-test/suite/period/t/fk.test
@@ -0,0 +1,261 @@
+--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');
+--sorted_result
+select * from t;
+--sorted_result
+select * from s;
+
+--error ER_ROW_IS_REFERENCED_2
+delete from t;
+
+--sorted_result
+select * from t;
+--sorted_result
+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');
+
+--sorted_result
+select * from t;
+--sorted_result
+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');
+
+--sorted_result
+select * from t;
+--sorted_result
+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;
+
+--sorted_result
+select * from t;
+--sorted_result
+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;
+
+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);
+delete from t;
+insert into t values(1, 1, '2017-01-03', '2017-01-20');
+
+show create table s;
+--sorted_result
+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;
+
+call mtr.add_suppression("In ALTER TABLE .* has or is referenced in foreign key constraints which are not compatible with the new table definition.");
+
+--replace_regex  /Error on rename of '.*' to '.*'/Error on rename of 'OLD_FILE_NAME' to 'NEW_FILE_NAME'/
+--error ER_ERROR_ON_RENAME
+alter table t drop period for p, drop constraint id;
+
+show create table s;
+alter table t drop period for p, drop constraint id, add period for pop(s, e),
+              add unique(id, x, pop without overlaps);
+show create table s;
+
+--replace_regex  /Error on rename of '.*' to '.*'/Error on rename of 'OLD_FILE_NAME' to 'NEW_FILE_NAME'/
+--error ER_ERROR_ON_RENAME
+alter table t drop period for pop, drop constraint id,
+              add unique(id, x, s, e);
+show create 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/mysql-test/suite/period/t/overlaps.test b/mysql-test/suite/period/t/overlaps.test
index 6cd78769d4a..5c946478d7c 100644
--- a/mysql-test/suite/period/t/overlaps.test
+++ b/mysql-test/suite/period/t/overlaps.test
@@ -5,6 +5,25 @@
 
 let $default_engine= `select @@default_storage_engine`;
 
+
+create or replace table t(id int, s date, e date,
+                          period for p(s,e));
+
+insert into t values (1, '2003-01-01', '2003-03-01'),
+                     (1, '2003-05-01', '2003-07-01'),
+                     (1, '2003-02-01', '2003-04-01');
+
+--replace_regex /#sql-\w+/#sql-temp/
+--error ER_DUP_ENTRY
+alter table t add primary key(id, p without overlaps);
+
+--echo # Historical rows are not checked against constraints
+set @@system_versioning_alter_history= keep;
+alter table t add system versioning;
+delete from t;
+alter table t add primary key(id, p without overlaps);
+
+
 create or replace table t(id int, s date, e date,
                           period for p(s,e),
                           primary key(id, p without overlaps)
diff --git a/mysql-test/suite/period/t/versioning.test b/mysql-test/suite/period/t/versioning.test
index 5d38d81349e..b607f948b7e 100644
--- a/mysql-test/suite/period/t/versioning.test
+++ b/mysql-test/suite/period/t/versioning.test
@@ -58,4 +58,23 @@ select *, if(row_start = @ins_time, "OLD", "NEW"), check_row(row_start, row_end)
 
 drop table t,log_tbl;
 drop procedure log;
+
+create or replace table t (id int, x int, s date, e date, period for p(s,e),
+                           unique(id, x, p without overlaps))
+                           with system versioning;
+
+--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, period p)
+                             on delete restrict);
+
+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) with system versioning;
+--error ER_DROP_INDEX_FK
+alter table t drop system versioning;
+alter table s drop system versioning;
+
 source suite/versioning/common_finish.inc;
diff --git a/sql/field.cc b/sql/field.cc
index 2619da5be26..58a4a846cb7 100644
--- a/sql/field.cc
+++ b/sql/field.cc
@@ -4606,17 +4606,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;
 }
 
@@ -5654,13 +5654,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,
diff --git a/sql/field.h b/sql/field.h
index 453f6af428b..fa82ce70778 100644
--- a/sql/field.h
+++ b/sql/field.h
@@ -776,8 +776,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
 
@@ -1118,7 +1123,7 @@ public:
   */
   virtual uint32 sort_suffix_length() const { return 0; }
 
-  /* 
+  /*
     Get the number bytes occupied by the value in the field.
     CHAR values are stripped of trailing spaces.
     Flexible values are stripped of their length.
@@ -1548,7 +1553,8 @@ public:
 
   uint get_key_image(uchar *buff, uint length, imagetype type_arg) const
   { return get_key_image(buff, length, ptr, type_arg); }
-  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;
@@ -2770,7 +2776,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;
@@ -3355,7 +3361,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;
   bool val_native(Native *to) override;
   uint size_of() const override { return sizeof *this; }
diff --git a/sql/handler.cc b/sql/handler.cc
index f223def4aa9..23f8dd86685 100644
--- a/sql/handler.cc
+++ b/sql/handler.cc
@@ -6986,6 +6986,35 @@ int handler::check_duplicate_long_entries_update(const uchar *new_rec)
   return 0;
 }
 
+static int period_locate_overlapping_record(handler *handler,
+                                            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;
+  const uint period_field_length= key_to_cmp.key_part[base_parts].length;
+  const uint key_base_length= key_to_cmp.key_length - 2 * period_field_length;
+
+
+  key_copy(key_buf, record_to_cmp, &key_to_cmp, 0);
+
+  /*
+    Copy period_start to period_end.
+    the value in period_start field is not significant, but anyway let's leave
+    it defined to avoid uninitialized memory access
+  */
+  memcpy(key_buf + key_base_length,
+         key_buf + key_base_length + period_field_length,
+         period_field_length);
+
+  int error= handler->ha_index_read_map(record, key_buf,
+                                        PREV_BITS(ulong, base_parts + 1),
+                                        HA_READ_AFTER_KEY);
+  return error;
+}
 
 int handler::ha_check_overlaps(const uchar *old_data, const uchar* new_data)
 {
@@ -7027,6 +7056,9 @@ int handler::ha_check_overlaps(const uchar *old_data, const uchar* new_data)
         continue;
     }
 
+    if (key_rec_is_null(key_info, new_data))
+      continue;
+
     error= lookup_handler->ha_index_init(key_nr, 0);
     if (error)
       return error;
@@ -7034,23 +7066,9 @@ int handler::ha_check_overlaps(const uchar *old_data, const uchar* new_data)
     error= lookup_handler->ha_start_keyread(key_nr);
     DBUG_ASSERT(!error);
 
-    const uint period_field_length= key_info.key_part[key_parts - 1].length;
-    const uint key_base_length= key_info.key_length - 2 * period_field_length;
-
-    key_copy(lookup_buffer, new_data, &key_info, 0);
-
-    /* Copy period_start to period_end.
-       the value in period_start field is not significant, but anyway let's leave
-       it defined to avoid uninitialized memory access
-     */
-    memcpy(lookup_buffer + key_base_length,
-           lookup_buffer + key_base_length + period_field_length,
-           period_field_length);
-
-    /* Find row with period_end > (period_start of new_data) */
-    error = lookup_handler->ha_index_read_map(record_buffer, lookup_buffer,
-                                       key_part_map((1 << (key_parts - 1)) - 1),
-                                       HA_READ_AFTER_KEY);
+    error= period_locate_overlapping_record(lookup_handler, lookup_buffer,
+                                            new_data, record_buffer,
+                                            key_info, key_info);
 
     if (!error && is_update)
     {
@@ -7183,6 +7201,10 @@ int handler::ha_write_row(const uchar *buf)
   if ((error= ha_check_overlaps(NULL, buf)))
     DBUG_RETURN(error);
 
+  error= period_row_ins_fk_check(buf);
+  if (unlikely(error))
+    DBUG_RETURN(error);
+
   if (table->s->long_unique_table && this == table->file)
   {
     DBUG_ASSERT(inited == NONE || lookup_handler != this);
@@ -7236,6 +7258,9 @@ int handler::ha_update_row(const uchar *old_data, const uchar *new_data)
   uint saved_status= table->status;
   error= ha_check_overlaps(old_data, new_data);
 
+  if (!error)
+    error= period_row_upd_fk_check(old_data, new_data);
+
   if (!error && table->s->long_unique_table && this == table->file)
     error= check_duplicate_long_entries_update(new_data);
   table->status= saved_status;
@@ -7297,6 +7322,277 @@ int handler::update_first_row(const uchar *new_data)
 }
 
 
+static int period_find_first_overlapping_record(handler *handler,
+                                                uchar *key_buf,
+                                                const uchar *record_to_cmp,
+                                                uchar *record,
+                                                const KEY &key_to_cmp,
+                                                const KEY &key)
+{
+  /*
+    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.
+
+    In ordinary btree, like now, since we have period_start, period_end at the
+    end of the key, we can find a row such that row
+    period_end > (period_start of record_to_cmp) and then scan until
+    period_start < (period_end of record_to_cmp) is satisfied.
+    This
+
+    Note that WITHOUT OVERLAPS on key makes it O(1).
+  */
+
+  int error= period_locate_overlapping_record(handler, key_buf,
+                                              record_to_cmp, record,
+                                              key_to_cmp, key);
+  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.empty())
+    return 0;
+  if (table->versioned() && !table->vers_end_field()->is_max(record))
+    return 0;
+
+  for(auto &fk: table->referenced)
+  {
+    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;
+
+    foreign_handler->alloc_lookup_buffer();
+
+    /*
+      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->lookup_buffer;
+
+    error= period_find_first_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,
+                                       uchar *key_buf,
+                                       const uchar *record,
+                                       uchar *ref_record,
+                                       const KEY &key,
+                                       const KEY &ref_key)
+{
+  int error= period_find_first_overlapping_record(ref_handler, key_buf,
+                                                  record, ref_record,
+                                                  key, ref_key);
+  if (error)
+    return error;
+
+  auto period_start= key.user_defined_key_parts - 1;
+  auto period_end= key.user_defined_key_parts - 2;
+
+  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.empty())
+    return 0;
+
+  for(auto &fk: table->foreign)
+  {
+    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;
+
+    ref_handler->alloc_lookup_buffer();
+
+    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->lookup_buffer;
+
+    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;
@@ -7308,6 +7604,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 ad6b4d54eac..9b7f7c67b86 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -4749,6 +4749,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/key.cc b/sql/key.cc
index ce0f637789c..7be6d8252d4 100644
--- a/sql/key.cc
+++ b/sql/key.cc
@@ -897,3 +897,70 @@ bool key_buf_cmp(KEY *key_info, uint used_key_parts,
   }
   return FALSE;
 }
+
+
+bool key_rec_is_null(const KEY &key, const uchar *rec)
+{
+  for (uint p= 0; p < key.user_defined_key_parts; p++)
+  {
+    KEY_PART_INFO &part= key.key_part[p];
+    if (part.field->is_null_in_record(rec))
+      return true;
+  }
+  return false;
+}
+
+int key_period_compare_bases(const KEY &lhs_key, const KEY &rhs_key,
+                             const uchar *lhs, const uchar *rhs)
+{
+  uint base_part_nr= lhs_key.user_defined_key_parts - 2;
+  int cmp_res= 0;
+  for (uint part_nr= 0; !cmp_res && part_nr < base_part_nr; part_nr++)
+  {
+    Field *fl= lhs_key.key_part[part_nr].field;
+    Field *fr= rhs_key.key_part[part_nr].field;
+
+    bool lhs_null= fl->is_null_in_record(lhs);
+    bool rhs_null= fr->is_null_in_record(rhs);
+    if (lhs_null || rhs_null)
+    {
+      if (lhs_null && rhs_null)
+        continue;
+      return lhs_null ? -1 : 1;
+    }
+
+    uint kp_len= MY_MIN(lhs_key.key_part[part_nr].length,
+                        rhs_key.key_part[part_nr].length);
+    cmp_res= fl->cmp_prefix(fl->ptr_in_record(lhs), fr->ptr_in_record(rhs),
+                            kp_len);
+  }
+
+  return cmp_res;
+}
+
+int key_period_compare_periods(const KEY &lhs_key, const KEY &rhs_key,
+                               const uchar *lhs, const uchar *rhs)
+{
+  uint base_part_nr= lhs_key.user_defined_key_parts - 2;
+
+  DBUG_ASSERT(!lhs_key.key_part[base_part_nr].null_bit);
+  DBUG_ASSERT(!lhs_key.key_part[base_part_nr + 1].null_bit);
+  DBUG_ASSERT(!rhs_key.key_part[base_part_nr].null_bit);
+  DBUG_ASSERT(!rhs_key.key_part[base_part_nr + 1].null_bit);
+
+  Field *lhs_fields[]= {lhs_key.key_part[base_part_nr + 1].field,
+                        lhs_key.key_part[base_part_nr].field};
+
+  Field *rhs_fields[]= {rhs_key.key_part[base_part_nr + 1].field,
+                        rhs_key.key_part[base_part_nr].field};
+
+  const Field *f= lhs_fields[0];
+
+  if (f->cmp(lhs_fields[1]->ptr_in_record(lhs),
+             rhs_fields[0]->ptr_in_record(rhs)) <= 0)
+    return -1;
+  if (f->cmp(lhs_fields[0]->ptr_in_record(lhs),
+             rhs_fields[1]->ptr_in_record(rhs)) >= 0)
+    return 1;
+  return 0;
+}
diff --git a/sql/key.h b/sql/key.h
index 871373bfcd5..98095fe7dd1 100644
--- a/sql/key.h
+++ b/sql/key.h
@@ -40,5 +40,10 @@ bool key_buf_cmp(KEY *key_info, uint used_key_parts,
                  const uchar *key1, const uchar *key2);
 extern "C" int key_rec_cmp(void *key_info, uchar *a, uchar *b);
 int key_tuple_cmp(KEY_PART_INFO *part, uchar *key1, uchar *key2, uint tuple_length);
+bool key_rec_is_null(const KEY &key, const uchar *rec);
+int key_period_compare_bases(const KEY &lhs_key, const KEY &rhs_key,
+                             const uchar *lhs, const uchar *rhs);
+int key_period_compare_periods(const KEY &lhs_key, const KEY &rhs_key,
+                               const uchar *lhs, const uchar *rhs);
 
 #endif /* KEY_INCLUDED */
diff --git a/sql/lex_string.h b/sql/lex_string.h
index 184788c3b88..3eced559698 100644
--- a/sql/lex_string.h
+++ b/sql/lex_string.h
@@ -110,6 +110,11 @@ class Lex_cstring : public LEX_CSTRING, public Sql_alloc
       return 1;
     return strcmp(str, rhs);
   }
+  bool defined() const
+  {
+    DBUG_ASSERT(MY_TEST(str) == MY_TEST(length));
+    return MY_TEST(str);
+  }
 };
 
 
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index f82adbc90d1..bbf219d3035 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -1804,7 +1804,7 @@ ER_WRONG_AUTO_KEY 42000 S1009
 ER_BINLOG_CANT_DELETE_GTID_DOMAIN
 	eng "Could not delete gtid domain. Reason: %s."
         ukr "Не можу видалити домен gtid. Причина: %s."
-ER_NORMAL_SHUTDOWN  
+ER_NORMAL_SHUTDOWN
         cze "%s (%s): normální ukončení"
         dan "%s (%s): Normal nedlukning"
         nla "%s (%s): Normaal afgesloten "
@@ -2205,7 +2205,7 @@ ER_UPDATE_TABLE_USED
         eng "Table '%-.192s' is specified twice, both as a target for '%s' and as a separate source for data"
         swe "Table '%-.192s' är använd två gånger. Både för '%s' och för att hämta data"
         ukr "Таблиця '%-.192s' вказується двічі, як цільова для '%s', так і як окреме джерело даних"
-ER_NO_SUCH_THREAD  
+ER_NO_SUCH_THREAD
         cze "Neznámá identifikace threadu: %lu"
         dan "Ukendt tråd id: %lu"
         nla "Onbekend thread id: %lu"
@@ -2827,7 +2827,7 @@ ER_NULL_COLUMN_IN_INDEX 42000
         eng "Table handler doesn't support NULL in given index. Please change column '%-.192s' to be NOT NULL or use another handler"
         swe "Tabell hanteraren kan inte indexera NULL kolumner för den givna index typen. Ändra '%-.192s' till NOT NULL eller använd en annan hanterare"
         ukr "Вказівник таблиці не підтримує NULL у зазначенному індексі. Будь ласка, зменіть стовпчик '%-.192s' на NOT NULL або використайте інший вказівник таблиці."
-ER_CANT_FIND_UDF  
+ER_CANT_FIND_UDF
         cze "Nemohu načíst funkci '%-.192s'"
         dan "Kan ikke læse funktionen '%-.192s'"
         nla "Kan functie '%-.192s' niet laden"
@@ -4483,7 +4483,7 @@ ER_UNION_TABLES_IN_DIFFERENT_DIR
         spa "Incorrecta definición de la tabla; Todas las tablas MERGE deben estar en el mismo banco de datos"
         swe "Felaktig tabelldefinition; alla tabeller i en MERGE-tabell måste vara i samma databas"
         ukr "Хибне визначення таблиці; всі MERGE-таблиці повинні належити до однієї бази ланних."
-ER_LOCK_DEADLOCK 40001 
+ER_LOCK_DEADLOCK 40001
         nla "Deadlock gevonden tijdens lock-aanvraag poging; Probeer herstart van de transactie"
         eng "Deadlock found when trying to get lock; try restarting transaction"
         est "Lukustamisel tekkis tupik (deadlock); alusta transaktsiooni otsast"
@@ -4497,7 +4497,7 @@ ER_LOCK_DEADLOCK 40001
         spa "Encontrado deadlock cuando tentando obtener el bloqueo; Tente recomenzar la transición"
         swe "Fick 'DEADLOCK' vid låsförsök av block/rad. Försök att starta om transaktionen"
         ukr "Взаємне блокування знайдено під час спроби отримати блокування; спробуйте перезапустити транзакцію."
-ER_TABLE_CANT_HANDLE_FT  
+ER_TABLE_CANT_HANDLE_FT
         nla "Het gebruikte tabel type (%s) ondersteund geen FULLTEXT indexen"
         eng "The storage engine %s doesn't support FULLTEXT indexes"
         est "Antud tabelitüüp (%s) ei toeta FULLTEXT indekseid"
@@ -4524,7 +4524,7 @@ ER_CANNOT_ADD_FOREIGN
         spa "No puede adicionar clave extranjera constraint para `%s`"
         swe "Kan inte lägga till 'FOREIGN KEY constraint' för `%s`'"
         ukr "Не можу додати обмеження зовнішнього ключа Ha `%s`"
-ER_NO_REFERENCED_ROW 23000 
+ER_NO_REFERENCED_ROW 23000
         nla "Kan onderliggende rij niet toevoegen: foreign key beperking gefaald"
         eng "Cannot add or update a child row: a foreign key constraint fails"
         fre "Impossible d'ajouter un enregistrement fils : une constrainte externe l'empèche"
@@ -4539,7 +4539,7 @@ ER_NO_REFERENCED_ROW 23000
         spa "No puede adicionar una línea hijo: falla de clave extranjera constraint"
         swe "FOREIGN KEY-konflikt:  Kan inte skriva barn"
         ukr "Не вдається додати або оновити дочірній рядок: невдала перевірка обмеження зовнішнього ключа"
-ER_ROW_IS_REFERENCED 23000 
+ER_ROW_IS_REFERENCED 23000
         eng "Cannot delete or update a parent row: a foreign key constraint fails"
         fre "Impossible de supprimer un enregistrement père : une constrainte externe l'empèche"
         ger "Löschen oder Aktualisieren eines Eltern-Datensatzes schlug aufgrund einer Fremdschlüssel-Beschränkung fehl"
@@ -7969,3 +7969,7 @@ ER_FUNCTION_CANNOT_BE_USED_IN_CLAUSE
         eng "Function '%s' cannot be used in the %s clause"
 ER_ORACLE_COMPAT_FUNCTION_ERROR
         eng "Oracle compatibility function error: %s"
+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"
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 14f02b903b7..f6cf57276a7 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -197,6 +197,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),
   fk_options(rhs.fk_options)
 {
   list_copy_and_replace_each_value(ref_columns, mem_root);
@@ -4979,6 +4980,11 @@ unsigned long long thd_get_query_id(const MYSQL_THD thd)
   return((unsigned long long)thd->query_id);
 }
 
+TABLE *thd_get_open_tables(const MYSQL_THD thd)
+{
+  return thd->open_tables;
+}
+
 void thd_clear_error(MYSQL_THD thd)
 {
   thd->clear_error();
diff --git a/sql/sql_class.h b/sql/sql_class.h
index 12fe0c36b23..89a983862af 100644
--- a/sql/sql_class.h
+++ b/sql/sql_class.h
@@ -675,7 +675,7 @@ public:
      type(type_par), foreign(false), generated(generated_arg),
      invisible(false), without_overlaps(false), ignore(false),
      key_create_info(default_key_create_info),
-     name(*name_arg), option_list(NULL)
+     name(*name_arg), option_list(NULL), period{}
   {
     key_create_info.algorithm= algorithm_arg;
   }
@@ -687,7 +687,7 @@ public:
      type(type_par), foreign(false), generated(generated_arg),
      invisible(false), without_overlaps(false), ignore(false),
      key_create_info(*key_info_arg), columns(*cols),
-     name(*name_arg), option_list(create_opt)
+     name(*name_arg), option_list(create_opt), period{}
   {}
   Key(const Key &rhs, MEM_ROOT *mem_root);
   virtual ~Key() {}
@@ -709,6 +709,7 @@ public:
   LEX_CSTRING ref_table;
   TABLE_LIST *ref_table_list;
   List<Key_part_spec> ref_columns;
+  Lex_ident ref_period;
   st_fk_options fk_options;
   Foreign_key(const LEX_CSTRING *name_arg,
               const LEX_CSTRING *constraint_name_arg,
@@ -740,6 +741,9 @@ public:
         return;
     }
 
+    period= src.period;
+    ref_period= src.ref_period;
+
     foreign= true; // false means failed initialization
   }
   bool failed() const
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index b28b856d7e3..74e1ec00582 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -2860,6 +2860,38 @@ mysql_prepare_create_table(THD *thd, HA_CREATE_INFO *create_info,
       bool self_ref;
       if (!fk_key->ignore && fk_key->validate(db, table_name, alter_info->create_list, self_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);
+        }
+      }
     }
     (*key_count)++;
     tmp=file->max_key_parts();
@@ -3867,10 +3899,29 @@ static int append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
   const Lex_ident &row_end_field= create_info->vers_info.as_row.end;
   DBUG_ASSERT(!create_info->versioned() || (row_start_field && row_end_field));
 
-  int result = 0;
+  int old_columns = key->columns.elements;
+
   if (create_info->versioned() && (key->type == Key::PRIMARY
                                    || key->type == Key::UNIQUE))
   {
+    if (!key->period && key->type != Key::PRIMARY && key->type != Key::UNIQUE)
+      return 0;
+
+    Foreign_key *fk = nullptr;
+    if (key->foreign)
+    {
+      fk= static_cast<Foreign_key *>(key);
+
+      if (fk->ref_table_list->table &&
+          fk->ref_table_list->table->versioned() != create_info->versioned())
+      {
+        my_error(ER_WRONG_FK_DEF, MYF(0), create_info->alias.str,
+                 create_info->versioned() ? "ADD SYSTEM VERSIONING"
+                                          : "DROP SYSTEM VERSIONING");
+        return -1;
+      }
+    }
+
     Key_part_spec *key_part=NULL;
     List_iterator<Key_part_spec> part_it(key->columns);
     while ((key_part=part_it++))
@@ -3879,18 +3930,26 @@ static int 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->foreign)
     {
-      key->columns.push_back(new (thd->mem_root)
-                               Key_part_spec(&row_end_field, 0, true));
-      result++;
+      key->columns.push_back(new Key_part_spec(&row_end_field, 0));
     }
 
+    if (key->foreign)
+    {
+      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));
+    }
   }
 
-  if (key->without_overlaps)
+  if (key->period)
   {
-    DBUG_ASSERT(key->type == Key::PRIMARY || key->type == Key::UNIQUE);
+    DBUG_ASSERT(key->without_overlaps || key->foreign);
+    DBUG_ASSERT(key->type == Key::PRIMARY || key->type == Key::UNIQUE
+                || key->foreign);
     if (!create_info->period_info.is_set()
         || !key->period.streq(create_info->period_info.name))
     {
@@ -3911,15 +3970,23 @@ static int append_system_key_parts(THD *thd, HA_CREATE_INFO *create_info,
         return -1;
       }
     }
-    const auto &period= create_info->period_info.period;
-    key->columns.push_back(new (thd->mem_root)
-                           Key_part_spec(&period.end, 0, true));
-    key->columns.push_back(new (thd->mem_root)
-                           Key_part_spec(&period.start, 0, true));
-    result += 2;
+
+    key->columns.push_back(new Key_part_spec(&period_end, 0));
+    key->columns.push_back(new Key_part_spec(&period_start, 0));
+
+    if (key->foreign)
+    {
+      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_end, 0));
+      fk->ref_columns.push_back(new Key_part_spec(&ref_period_start, 0));
+    }
   }
 
-  return result;
+  return key->columns.elements - old_columns;
 }
 
 handler *mysql_create_frm_image(THD *thd, const LEX_CSTRING &db,
@@ -8337,8 +8404,8 @@ mysql_prepare_alter_table(THD *thd, TABLE *table,
       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->without_overlaps= key_info->without_overlaps;
       new_key_list.push_back(key, thd->mem_root);
     }
     if (long_hash_key)
@@ -10089,6 +10156,11 @@ do_continue:;
         }
       }
     }
+    for (uint i= 0; i < table->s->keys; i++)
+    {
+      if (table->key_info[i].without_overlaps)
+        ha_alter_info.inplace_supported= HA_ALTER_INPLACE_NOT_SUPPORTED;
+    }
 
     if (alter_info->supports_algorithm(thd, &ha_alter_info) ||
         alter_info->supports_lock(thd, &ha_alter_info))
diff --git a/sql/sql_type.cc b/sql/sql_type.cc
index cac2ef1d354..827b02e9ed3 100644
--- a/sql/sql_type.cc
+++ b/sql/sql_type.cc
@@ -1503,15 +1503,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 68959e1a11a..99a6896cda2 100644
--- a/sql/sql_type.h
+++ b/sql/sql_type.h
@@ -6100,7 +6100,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() {}
@@ -6403,9 +6403,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
@@ -6576,7 +6577,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 b6e34234217..af8d2d5c21e 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -5921,7 +5921,7 @@ key_def:
                MYSQL_YYABORT;
             Lex->option_list= NULL;
           }
-          '(' key_list ')' references
+          '(' key_list_fk ')' references
           {
             if (unlikely(Lex->init_last_foreign_key($10)))
               MYSQL_YYABORT;
@@ -6873,6 +6873,10 @@ ref_list:
               MYSQL_YYABORT;
             Lex->last_foreign_key->ref_columns.push_back(key, thd->mem_root);
           }
+        | ref_list ',' PERIOD_SYM ident
+          {
+            Lex->last_foreign_key->ref_period= $4;
+          }
         | ident
           {
             Key_part_spec *key= new (thd->mem_root) Key_part_spec(&$1, 0);
@@ -7091,6 +7095,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.h b/sql/table.h
index 9ce4786d2eb..d974c359da2 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1257,6 +1257,7 @@ struct st_cond_statistic;
 class SplM_opt_info;
 
 struct vers_select_conds_t;
+struct FOREIGN_KEY;
 
 struct TABLE
 {
@@ -1318,6 +1319,9 @@ public:
   Field *found_next_number_field;	/* Set on open */
   Virtual_column_info **check_constraints;
 
+  st_::span<FOREIGN_KEY> foreign;
+  st_::span<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 */
@@ -1845,7 +1849,8 @@ enum enum_schema_table_state
 };
 
 enum enum_fk_option { FK_OPTION_UNDEF= 0, FK_OPTION_RESTRICT, FK_OPTION_CASCADE,
-               FK_OPTION_SET_NULL, FK_OPTION_NO_ACTION, FK_OPTION_SET_DEFAULT};
+               FK_OPTION_SET_NULL, FK_OPTION_NO_ACTION, FK_OPTION_SET_DEFAULT,
+               FK_OPTION_LAST= FK_OPTION_SET_DEFAULT};
 
 enum fk_match_opt { FK_MATCH_UNDEF, FK_MATCH_FULL,
   FK_MATCH_PARTIAL, FK_MATCH_SIMPLE};
@@ -1876,6 +1881,8 @@ public:
   Lex_cstring referenced_table;
   st_::span<Lex_cstring> foreign_fields;
   st_::span<Lex_cstring> referenced_fields;
+  Lex_cstring period;
+  Lex_cstring ref_period;
   enum_fk_option update_method;
   enum_fk_option delete_method;
 
@@ -1933,6 +1940,18 @@ typedef class FK_info 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;
 
 /*
@@ -2697,7 +2716,7 @@ struct TABLE_LIST
   }
   void print(THD *thd, table_map eliminated_tables, String *str, 
              enum_query_type query_type);
-  bool check_single_table(TABLE_LIST **table, table_map map,
+  bool check_single_table(TABLE_LIST **table, table_map map_arg,
                           TABLE_LIST *view);
   bool set_insert_values(MEM_ROOT *mem_root);
   void hide_view_error(THD *thd);
diff --git a/sql/unireg.cc b/sql/unireg.cc
index 464cc9d8293..1861c8c8bfd 100644
--- a/sql/unireg.cc
+++ b/sql/unireg.cc
@@ -1248,6 +1248,9 @@ ulonglong Foreign_key_io::fk_size(FK_info &fk)
     store_size+= string_size(fk.foreign_fields[i]);
     store_size+= string_size(fk.referenced_fields[i]);
   }
+  if (fk.period.defined())
+    store_size+= string_size(fk.period) + string_size(fk.ref_period);
+
   return store_size;
 }
 
@@ -1279,6 +1282,9 @@ void Foreign_key_io::store_fk(FK_info &fk, uchar *&pos)
     pos= store_string(pos, fk.referenced_db, true);
     pos= store_string(pos, fk.referenced_table);
   }
+  static_assert(PERIOD_MASK > FK_OPTION_LAST,
+                "period bit is stored as bit 4 in the format");
+  int period_flag = fk.period.defined() << PERIOD_BIT;
   pos= store_length(pos, fk.update_method);
   pos= store_length(pos, fk.delete_method);
   pos= store_length(pos, fk.foreign_fields.size());
@@ -1288,6 +1294,11 @@ void Foreign_key_io::store_fk(FK_info &fk, uchar *&pos)
     pos= store_string(pos, fk.foreign_fields[i]);
     pos= store_string(pos, fk.referenced_fields[i]);
   }
+  if (fk.period.defined())
+  {
+    pos= store_string(pos, fk.period);
+    pos= store_string(pos, fk.ref_period);
+  }
   DBUG_ASSERT(pos - old_pos == (long int)fk_size(fk));
 }
 
@@ -1410,7 +1421,8 @@ bool Foreign_key_io::parse(THD *thd, TABLE_SHARE *s, LEX_CUSTRING& image)
       return true;
     if (update_method > FK_OPTION_SET_DEFAULT || delete_method > FK_OPTION_SET_DEFAULT)
       return true;
-    dst->update_method= (enum_fk_option) update_method;
+    bool has_period= update_method & PERIOD_MASK;
+    dst->update_method= (enum_fk_option) (update_method & ~PERIOD_MASK);
     dst->delete_method= (enum_fk_option) delete_method;
     size_t col_count;
     if (read_length(col_count, p))
@@ -1427,6 +1439,13 @@ bool Foreign_key_io::parse(THD *thd, TABLE_SHARE *s, LEX_CUSTRING& image)
       if (read_string(dst->referenced_fields[j], &s->mem_root, p))
         return true;
     }
+    if (has_period)
+    {
+      if (read_string(dst->period, &s->mem_root, p))
+        return true;
+      if (read_string(dst->ref_period, &s->mem_root, p))
+        return true;
+    }
     /* If it is self-reference we also push to referenced_keys: */
     if (dst->self_ref() && s->referenced_keys.push_back(dst, &s->mem_root))
     {
@@ -1589,6 +1608,16 @@ bool TABLE_SHARE::fk_resolve_referenced_keys(THD *thd, TABLE_SHARE *from)
                             fld.str, from->db.str, from->table_name.str);
         return true;
       }
+      if (fk.ref_period.defined() && fk.ref_period.cmp(period.name) != 0)
+      {
+        push_warning_printf(thd, Sql_condition::WARN_LEVEL_WARN,
+                            ER_CANNOT_ADD_FOREIGN,
+                            "Period `%s` hint table `%s.%s` refers to"
+                            " does not exist",
+                            fk.ref_period.str, from->db.str,
+                            from->table_name.str);
+        return true;
+      }
     }
     FK_info *dst= fk.clone(&mem_root);
     if (referenced_keys.push_back(dst, &mem_root))
diff --git a/sql/unireg.h b/sql/unireg.h
index ced4b54a9d6..1bb4011cbec 100644
--- a/sql/unireg.h
+++ b/sql/unireg.h
@@ -182,6 +182,8 @@ class Foreign_key_io: public BinaryStringBuffer<512>
 {
 public:
   static const ulonglong fk_io_version= 0;
+  static constexpr uint PERIOD_BIT = 4;
+  static constexpr uint PERIOD_MASK = 1 << PERIOD_BIT;
   struct Pos
   {
     uchar *pos;
diff --git a/storage/innobase/dict/dict0dict.cc b/storage/innobase/dict/dict0dict.cc
index 626df54a221..5193fe970db 100644
--- a/storage/innobase/dict/dict0dict.cc
+++ b/storage/innobase/dict/dict0dict.cc
@@ -2951,6 +2951,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
@@ -2974,7 +2978,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;
@@ -3093,7 +3097,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, for_in_cache->has_period);
 
 		if (index == NULL
 		    && !(ignore_err & DICT_ERR_IGNORE_FK_NOKEY)) {
@@ -3132,7 +3136,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),
+                        for_in_cache->has_period);
 
 		if (index == NULL
 		    && !(ignore_err & DICT_ERR_IGNORE_FK_NOKEY)) {
@@ -4065,13 +4070,16 @@ dict_index_calc_min_rec_len(
 	return(sum);
 }
 
+TABLE *thd_get_open_tables(const MYSQL_THD thd);
+TABLE *find_locked_table(TABLE *list, const char *db, const char *table_name);
+
 /**********************************************************************//**
 Outputs info on a foreign key of a table in a format suitable for
 CREATE TABLE. */
 std::string
 dict_print_info_on_foreign_key_in_create_format(
 /*============================================*/
-	trx_t*		trx,		/*!< in: transaction */
+	trx_t*		trx,		/*!< in: transaction, can be NULL */
 	dict_foreign_t*	foreign,	/*!< in: foreign key constraint */
 	ibool		add_newline)	/*!< in: whether to add a newline */
 {
@@ -4101,16 +4109,63 @@ dict_print_info_on_foreign_key_in_create_format(
 	str.append(innobase_quote_identifier(trx, stripped_id));
 	str.append(" FOREIGN KEY (");
 
+	const char *period_name = NULL, *ref_period_name = NULL;
+	ulint n_fields = foreign->n_fields;
+
+	if (foreign->has_period)
+	{
+		n_fields -= 2;
+		if (foreign->foreign_table->versioned())
+			n_fields--;
+
+		if (!trx)
+		{
+			return "ERROR (period)";
+		}
+
+		char	db_buf[NAME_LEN + 1];
+		char	tbl_buf[NAME_LEN + 1];
+		ulint	db_buf_len, tbl_buf_len;
+
+		if (!foreign->foreign_table->parse_name<true>(
+		        db_buf, tbl_buf, &db_buf_len, &tbl_buf_len)) {
+			ut_ad(false);
+			return "ERROR (period)";
+		}
+		THD * thd = trx->mysql_thd;
+		TABLE *mysql_table = find_locked_table(
+						thd_get_open_tables(thd),
+						db_buf, tbl_buf);
+		DBUG_ASSERT(mysql_table);
+		period_name = mysql_table->s->period.name.str;
+
+		if (!foreign->referenced_table->parse_name<true>(
+		        db_buf, tbl_buf, &db_buf_len, &tbl_buf_len)) {
+			ut_ad(false);
+			return "ERROR (period)";
+		}
+		mysql_table = find_locked_table(thd_get_open_tables(thd),
+						db_buf, tbl_buf);
+		DBUG_ASSERT(mysql_table);
+		ref_period_name = mysql_table->s->period.name.str;
+	}
+
 	for (i = 0;;) {
 		str.append(innobase_quote_identifier(trx, foreign->foreign_col_names[i]));
 
-		if (++i < foreign->n_fields) {
+		if (++i < n_fields) {
 			str.append(", ");
 		} else {
 			break;
 		}
 	}
 
+	if (foreign->has_period)
+	{
+		str.append(", period ");
+		str.append(innobase_quote_identifier(trx, period_name));
+	}
+
 	str.append(") REFERENCES ");
 
 	if (dict_tables_have_same_db(foreign->foreign_table_name_lookup,
@@ -4130,13 +4185,19 @@ dict_print_info_on_foreign_key_in_create_format(
 		str.append(innobase_quote_identifier(trx,
 				foreign->referenced_col_names[i]));
 
-		if (++i < foreign->n_fields) {
+		if (++i < n_fields) {
 			str.append(", ");
 		} else {
 			break;
 		}
 	}
 
+	if (foreign->has_period)
+	{
+		str.append(", period ");
+		str.append(innobase_quote_identifier(trx, ref_period_name));
+	}
+
 	str.append(")");
 
 	if (foreign->type & DICT_FOREIGN_ON_DELETE_CASCADE) {
@@ -4610,7 +4671,7 @@ dict_foreign_replace_index(
 				foreign->foreign_col_names,
 				foreign->n_fields, index,
 				/*check_charsets=*/TRUE, /*check_null=*/FALSE,
-				NULL, NULL, NULL);
+				foreign->has_period, NULL, NULL, NULL);
 			if (new_index) {
 				ut_ad(new_index->table == index->table);
 				ut_ad(!new_index->to_be_dropped);
@@ -4635,7 +4696,7 @@ dict_foreign_replace_index(
 				foreign->referenced_col_names,
 				foreign->n_fields, index,
 				/*check_charsets=*/TRUE, /*check_null=*/FALSE,
-				NULL, NULL, NULL);
+				foreign->has_period, NULL, NULL, NULL);
 			/* There must exist an alternative index,
 			since this must have been checked earlier. */
 			if (new_index) {
@@ -5132,6 +5193,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
@@ -5144,6 +5209,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 - 1);
+		auto pend = dict_index_get_nth_field(index, n_cols - 2);
+		if ((pstart->col->prtype & DATA_PERIOD_START) == 0
+		    || (pend->col->prtype & DATA_PERIOD_END) == 0) {
+			return false;
+		}
+	}
+
 	if (index->type & (DICT_SPATIAL | DICT_FTS | DICT_CORRUPT)) {
 		return false;
 	}
diff --git a/storage/innobase/dict/dict0load.cc b/storage/innobase/dict/dict0load.cc
index 9c416138d84..dc919fb7755 100644
--- a/storage/innobase/dict/dict0load.cc
+++ b/storage/innobase/dict/dict0load.cc
@@ -2122,7 +2122,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
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index c9d86139741..26d8c78fcea 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -347,6 +347,13 @@ dict_mem_table_add_col(
 		ut_ad(!table->vers_end);
 		table->vers_end = i & dict_index_t::MAX_N_FIELDS;
 	}
+	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.
@@ -611,7 +618,7 @@ dict_mem_table_col_rename_low(
 				foreign->foreign_table, NULL,
 				foreign->foreign_col_names,
 				foreign->n_fields, NULL, true, false,
-				NULL, NULL, NULL);
+				foreign->has_period, NULL, NULL, NULL);
 
 		} else {
 
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 879de299914..b9d375b1f5c 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -9505,7 +9505,7 @@ wsrep_append_foreign_key(
 						foreign->referenced_col_names,
 						foreign->n_fields,
 						foreign->foreign_index,
-						TRUE, FALSE);
+						true, false, false);
 			}
 		} else {
 	  		foreign->foreign_table =
@@ -9519,7 +9519,7 @@ wsrep_append_foreign_key(
 						foreign->foreign_col_names,
 						foreign->n_fields,
 						foreign->referenced_index,
-						TRUE, FALSE);
+						true, false, false);
 			}
 		}
 		dict_sys.mutex_unlock();
@@ -10205,18 +10205,25 @@ 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;
 				ut_d(have_vers_start = true);
 			} else if (i == m_form->s->vers.end_fieldno) {
-				vers_row = DATA_VERS_END;
+				vers_period_row = DATA_VERS_END;
 				ut_d(have_vers_end = true);
 			} 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;
 			}
 		}
 
@@ -10306,7 +10313,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) {
@@ -10316,7 +10323,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);
@@ -10476,6 +10483,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,
@@ -10511,6 +10519,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));
@@ -11778,6 +11791,30 @@ public:
 	const char* str() { return buf; }
 };
 
+static const char *extract_fk_column(dict_table_t *table, trx_t *trx,
+				 dict_foreign_t *foreign, FK_info *fk,
+                                 const Lex_cstring &col,
+				 char *create_name,
+				 const char *operation,
+				 int ncol)
+{
+	const char *column_name= mem_heap_strdupl(foreign->heap,
+						  col.str,
+						  col.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;
+	}
+	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 */
@@ -11878,8 +11915,6 @@ create_table_info_t::create_foreign_keys()
 			continue;
 		}
 
-		bool	       success;
-
 		dict_foreign_t* foreign = dict_mem_foreign_create();
 		if (!foreign) {
 			return (DB_OUT_OF_MEMORY);
@@ -11924,40 +11959,29 @@ create_table_info_t::create_foreign_keys()
 			return (DB_CANNOT_ADD_CONSTRAINT);
 		}
 
+		if (fk->foreign_fields.size() >= MAX_NUM_FK_COLUMNS) {
+			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(),
+				fk->foreign_fields.size(),
+				MAX_NUM_FK_COLUMNS);
+			return DB_CANNOT_ADD_CONSTRAINT;
+		}
 
 		unsigned			  i = 0, j = 0;
 		for (Lex_cstring &col: fk->foreign_fields) {
-			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_NUM_FK_COLUMNS) {
-				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_NUM_FK_COLUMNS);
-				return (DB_CANNOT_ADD_CONSTRAINT);
-			}
+			column_names[i] = extract_fk_column(table, m_trx,
+				foreign, fk, col, create_name, operation, i);
+			if (!column_names[i])
+				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) {
@@ -11980,6 +12004,7 @@ create_table_info_t::create_foreign_keys()
 
 		foreign->foreign_index = index;
 		foreign->n_fields      = i & dict_index_t::MAX_N_FIELDS;
+		foreign->has_period    = fk->period.defined();
 
 		foreign->foreign_col_names = static_cast<const char**>(
 			mem_heap_alloc(foreign->heap, i * sizeof(void*)));
@@ -12030,26 +12055,11 @@ create_table_info_t::create_foreign_keys()
 		}
 
 		for (Lex_cstring &col: fk->referenced_fields) {
-			ref_column_names[j] = mem_heap_strdupl(
-				foreign->heap, col.str, col.length);
-			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);
-				}
-			}
+			ref_column_names[j] = extract_fk_column(
+				foreign->referenced_table, m_trx, foreign, fk,
+				col, create_name, operation, j);
+			if (!ref_column_names[j])
+				return DB_CANNOT_ADD_CONSTRAINT;
 			++j;
 		}
 		/* See ER_WRONG_FK_DEF in mysql_prepare_create_table() */
@@ -12063,8 +12073,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->period.defined(),
+				&index_error, &err_col, &err_index);
 
 			if (!index) {
 				key_text k(fk);
@@ -20953,13 +20963,13 @@ fk_upgrade_push_fk(
 	const char*			ref_column_names[MAX_NUM_FK_COLUMNS];
 	dict_index_t*			index;
 	for (uint i = 0; i < fk.foreign_fields.size(); i++) {
-		column_names[i]	      = fk.foreign_fields[i].str;
+		column_names[i]	    = fk.foreign_fields[i].str;
 		ref_column_names[i] = fk.referenced_fields[i].str;
 	}
 	dict_sys.mutex_lock();
 	index = dict_foreign_find_index(d.table, NULL, column_names,
 					fk.foreign_fields.size(), NULL, true,
-					false);
+					false, false);
 	if (!index) {
 		dict_sys.mutex_unlock();
 		ib_foreign_warn(
@@ -20985,7 +20995,7 @@ fk_upgrade_push_fk(
 	}
 	index = dict_foreign_find_index(ref_table, NULL, ref_column_names,
 					fk.foreign_fields.size(), NULL, true,
-					false);
+					false, false);
 	dict_table_close(ref_table, true, false);
 	dict_sys.mutex_unlock();
 	if (!index) {
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 75350f4a69c..2be3aca88ed 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -510,6 +510,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));
@@ -2589,8 +2590,10 @@ innobase_init_foreign(
 	dict_index_t*	referenced_index,	/*!< in: referenced index */
 	const char**	referenced_column_names,/*!< in: referenced column
 						names */
-	ulint		referenced_num_field)	/*!< in: number of referenced
+	ulint		referenced_num_field,	/*!< in: number of referenced
 						columns */
+	bool has_period)			/*!< in: has Application-time
+						period reference */
 {
 	dict_sys.assert_locked();
 
@@ -2628,6 +2631,7 @@ innobase_init_foreign(
         foreign->foreign_index = index;
         foreign->n_fields = static_cast<unsigned>(num_field)
 		& dict_index_t::MAX_N_FIELDS;
+        foreign->has_period = has_period;
 
         foreign->foreign_col_names = static_cast<const char**>(
                 mem_heap_alloc(foreign->heap, num_field * sizeof(void*)));
@@ -2824,7 +2828,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;
@@ -3035,6 +3039,7 @@ innobase_get_foreign_key_info(
 						referenced_column_names,
 						i, index,
 						TRUE, FALSE,
+						fk_key.ref_period.defined(),
 						NULL, NULL, NULL);
 
 				DBUG_EXECUTE_IF(
@@ -3069,7 +3074,8 @@ innobase_get_foreign_key_info(
 			    table, index, column_names,
 			    num_col, referenced_table_name,
 			    referenced_table, referenced_index,
-			    referenced_column_names, referenced_num_col)) {
+			    referenced_column_names, referenced_num_col,
+			    fk_key.period.defined())) {
 			dict_sys.mutex_unlock();
 			my_error(
 				ER_DUP_CONSTRAINT_NAME,
@@ -7286,7 +7292,7 @@ innobase_check_foreign_key_index(
 			    foreign->n_fields, index,
 			    /*check_charsets=*/TRUE,
 			    /*check_null=*/FALSE,
-			    NULL, NULL, NULL)
+			    foreign->has_period, NULL, NULL, NULL)
 		    && NULL == innobase_find_equiv_index(
 			    foreign->referenced_col_names,
 			    foreign->n_fields,
@@ -7337,7 +7343,7 @@ innobase_check_foreign_key_index(
 			    foreign->n_fields, index,
 			    /*check_charsets=*/TRUE,
 			    /*check_null=*/FALSE,
-			    NULL, NULL, NULL)
+			    foreign->has_period, NULL, NULL, NULL)
 		    && NULL == innobase_find_equiv_index(
 			    foreign->foreign_col_names,
 			    foreign->n_fields,
@@ -9579,7 +9585,7 @@ innobase_update_foreign_try(
 				fk->type
 				& (DICT_FOREIGN_ON_DELETE_SET_NULL
 					| DICT_FOREIGN_ON_UPDATE_SET_NULL),
-				NULL, NULL, NULL);
+				fk->has_period, 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 750c35341d6..38639be82a4 100644
--- a/storage/innobase/include/data0type.h
+++ b/storage/innobase/include/data0type.h
@@ -196,6 +196,12 @@ constexpr uint8_t DATA_MBR_LEN= uint8_t(SPDIMS * 2 * sizeof(double));
 /** 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
 
@@ -354,7 +360,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 401eb571ca9..355b0b88428 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -541,6 +541,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
@@ -621,6 +625,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 45820f6f3c7..e49ae7445da 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -73,7 +73,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 */
 /* @} */
 
@@ -728,6 +731,7 @@ public:
     return same_type(other) && len >= other.len &&
       mbminlen == other.mbminlen && 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));
   }
@@ -1419,6 +1423,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 */
@@ -2093,6 +2102,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 78a1da92736..0ff515ce367 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -1489,6 +1489,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 */