MDEV-16973 Application-time periods: DELETE

* inject portion of time updates into mysql_delete main loop
* triggered case emits delete+insert, no updates
* PORTION OF `SYSTEM_TIME` is forbidden
* `DELETE HISTORY .. FOR PORTION OF ...` is forbidden as well

15 files changed, 1048 insertions, 128 deletions

diff --git a/mysql-test/suite/period/create_triggers.inc b/mysql-test/suite/period/create_triggers.inc
new file mode 100644
index 00000000000..7263c5de7a4
--- /dev/null
+++ b/mysql-test/suite/period/create_triggers.inc
@@ -0,0 +1,41 @@
+disable_query_log;
+
+create or replace table log_tbl(id int auto_increment primary key, log text);
+
+create or replace procedure log(s text)
+  insert into log_tbl(log) values(s);
+
+if (!$trig_table)
+{
+  die "No $trig_table specified";
+}
+
+if (!$trig_cols)
+{
+  let $trig_cols= s, e;
+}
+
+let $old_trig_args= `select REGEXP_REPLACE('$trig_cols', '([[:word:]]+)',
+                                           'old.\\\\\\\\1')`;
+let $old_trig_args= `select REPLACE('$old_trig_args', ',', ', ", ", ')`;
+let $new_trig_args= `select REGEXP_REPLACE('$trig_cols', '([[:word:]]+)',
+                                           'new.\\\\\\\\1')`;
+let $new_trig_args= `select REPLACE('$new_trig_args', ',', ', ", ", ')`;
+
+eval create trigger tr1upd_$trig_table before update on $trig_table
+  for each row call log(CONCAT('>UPD: ', $old_trig_args, ' -> ', $new_trig_args));
+eval create trigger tr2upd_$trig_table after  update on $trig_table
+  for each row call log(CONCAT('<UPD: ', $old_trig_args, ' -> ', $new_trig_args));
+eval create trigger tr1del_$trig_table before delete on $trig_table
+  for each row call log(CONCAT('>DEL: ', $old_trig_args));
+eval create trigger tr2del_$trig_table after  delete on $trig_table
+  for each row call log(CONCAT('<DEL: ', $old_trig_args));
+eval create trigger tr1ins_$trig_table before insert on $trig_table
+  for each row call log(CONCAT('>INS: ', $new_trig_args));
+eval create trigger tr2ins_$trig_table after  insert on $trig_table
+  for each row call log(CONCAT('<INS: ', $new_trig_args));
+
+
+let trig_cols= 0;
+let trig_table= 0;
+enable_query_log;
diff --git a/mysql-test/suite/period/r/delete.result b/mysql-test/suite/period/r/delete.result
new file mode 100644
index 00000000000..f34a89ae2f0
--- /dev/null
+++ b/mysql-test/suite/period/r/delete.result
@@ -0,0 +1,299 @@
+create or replace table t (id int, s date, e date, period for apptime(s,e));
+insert into t values(1, '1999-01-01', '2018-12-12');
+insert into t values(1, '1999-01-01', '2017-01-01');
+insert into t values(1, '2017-01-01', '2019-01-01');
+insert into t values(2, '1998-01-01', '2018-12-12');
+insert into t values(3, '1997-01-01', '2015-01-01');
+insert into t values(4, '2016-01-01', '2020-01-01');
+insert into t values(5, '2010-01-01', '2015-01-01');
+create or replace table t1 (id int, s date, e date, period for apptime(s,e));
+insert t1 select * from t;
+create or replace table t2 (id int, s date, e date, period for apptime(s,e));
+insert t2 select * from t;
+create or replace table t3 (id int, s date, e date, period for apptime(s,e));
+insert t3 select * from t;
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+delete from t1 for portion of APPTIME from '2000-01-01' to '2018-01-01';
+select * from t;
+id	s	e
+1	1999-01-01	2000-01-01
+1	1999-01-01	2000-01-01
+1	2018-01-01	2018-12-12
+1	2018-01-01	2019-01-01
+2	1998-01-01	2000-01-01
+2	2018-01-01	2018-12-12
+3	1997-01-01	2000-01-01
+4	2018-01-01	2020-01-01
+select * from t1;
+id	s	e
+1	1999-01-01	2000-01-01
+1	1999-01-01	2000-01-01
+1	2018-01-01	2018-12-12
+1	2018-01-01	2019-01-01
+2	1998-01-01	2000-01-01
+2	2018-01-01	2018-12-12
+3	1997-01-01	2000-01-01
+4	2018-01-01	2020-01-01
+select * from log_tbl order by id;
+id	log
+1	>DEL: 1, 1999-01-01, 2018-12-12
+2	>INS: 1, 1999-01-01, 2000-01-01
+3	<INS: 1, 1999-01-01, 2000-01-01
+4	>INS: 1, 2018-01-01, 2018-12-12
+5	<INS: 1, 2018-01-01, 2018-12-12
+6	<DEL: 1, 1999-01-01, 2018-12-12
+7	>DEL: 1, 1999-01-01, 2017-01-01
+8	>INS: 1, 1999-01-01, 2000-01-01
+9	<INS: 1, 1999-01-01, 2000-01-01
+10	<DEL: 1, 1999-01-01, 2017-01-01
+11	>DEL: 1, 2017-01-01, 2019-01-01
+12	>INS: 1, 2018-01-01, 2019-01-01
+13	<INS: 1, 2018-01-01, 2019-01-01
+14	<DEL: 1, 2017-01-01, 2019-01-01
+15	>DEL: 2, 1998-01-01, 2018-12-12
+16	>INS: 2, 1998-01-01, 2000-01-01
+17	<INS: 2, 1998-01-01, 2000-01-01
+18	>INS: 2, 2018-01-01, 2018-12-12
+19	<INS: 2, 2018-01-01, 2018-12-12
+20	<DEL: 2, 1998-01-01, 2018-12-12
+21	>DEL: 3, 1997-01-01, 2015-01-01
+22	>INS: 3, 1997-01-01, 2000-01-01
+23	<INS: 3, 1997-01-01, 2000-01-01
+24	<DEL: 3, 1997-01-01, 2015-01-01
+25	>DEL: 4, 2016-01-01, 2020-01-01
+26	>INS: 4, 2018-01-01, 2020-01-01
+27	<INS: 4, 2018-01-01, 2020-01-01
+28	<DEL: 4, 2016-01-01, 2020-01-01
+29	>DEL: 5, 2010-01-01, 2015-01-01
+30	<DEL: 5, 2010-01-01, 2015-01-01
+# INSERT trigger only also works
+drop trigger tr1del_t2;
+drop trigger tr2del_t2;
+delete from t2 for portion of APPTIME from '2000-01-01' to '2018-01-01';
+select * from log_tbl order by id;
+id	log
+1	>INS: 1, 1999-01-01, 2000-01-01
+2	<INS: 1, 1999-01-01, 2000-01-01
+3	>INS: 1, 2018-01-01, 2018-12-12
+4	<INS: 1, 2018-01-01, 2018-12-12
+5	>INS: 1, 1999-01-01, 2000-01-01
+6	<INS: 1, 1999-01-01, 2000-01-01
+7	>INS: 1, 2018-01-01, 2019-01-01
+8	<INS: 1, 2018-01-01, 2019-01-01
+9	>INS: 2, 1998-01-01, 2000-01-01
+10	<INS: 2, 1998-01-01, 2000-01-01
+11	>INS: 2, 2018-01-01, 2018-12-12
+12	<INS: 2, 2018-01-01, 2018-12-12
+13	>INS: 3, 1997-01-01, 2000-01-01
+14	<INS: 3, 1997-01-01, 2000-01-01
+15	>INS: 4, 2018-01-01, 2020-01-01
+16	<INS: 4, 2018-01-01, 2020-01-01
+# removing BEFORE INSERT trigger enables internal substitution
+# DELETE+INSERT -> UPDATE, but without any side effects.
+# The optimization is disabled for non-transactional engines
+drop trigger tr1ins_t3;
+delete from t3 for portion of APPTIME from '2000-01-01' to '2018-01-01';
+select * from log_tbl order by id;
+id	log
+1	>DEL: 1999-01-01, 2018-12-12
+2	<INS: 1999-01-01, 2000-01-01
+3	<INS: 2018-01-01, 2018-12-12
+4	<DEL: 1999-01-01, 2018-12-12
+5	>DEL: 1999-01-01, 2017-01-01
+6	<INS: 1999-01-01, 2000-01-01
+7	<DEL: 1999-01-01, 2017-01-01
+8	>DEL: 2017-01-01, 2019-01-01
+9	<INS: 2018-01-01, 2019-01-01
+10	<DEL: 2017-01-01, 2019-01-01
+11	>DEL: 1998-01-01, 2018-12-12
+12	<INS: 1998-01-01, 2000-01-01
+13	<INS: 2018-01-01, 2018-12-12
+14	<DEL: 1998-01-01, 2018-12-12
+15	>DEL: 1997-01-01, 2015-01-01
+16	<INS: 1997-01-01, 2000-01-01
+17	<DEL: 1997-01-01, 2015-01-01
+18	>DEL: 2016-01-01, 2020-01-01
+19	<INS: 2018-01-01, 2020-01-01
+20	<DEL: 2016-01-01, 2020-01-01
+21	>DEL: 2010-01-01, 2015-01-01
+22	<DEL: 2010-01-01, 2015-01-01
+# multi-table DELETE is not possible
+delete t, t1 from t1, t for portion of apptime from '2000-01-01' to '2018-01-01';
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'for portion of apptime from '2000-01-01' to '2018-01-01'' at line 1
+delete t for portion of apptime from '2000-01-01' to '2018-01-01', t1 from t, t1;
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'for portion of apptime from '2000-01-01' to '2018-01-01', t1 from t, t1' at line 1
+# Here another check fails before parsing ends
+delete t, t1 from t for portion of apptime from '2000-01-01' to '2018-01-01', t1;
+ERROR 42S02: Unknown table 't1' in MULTI DELETE
+delete history from t2 for portion of apptime from '2000-01-01' to '2018-01-01';
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'for portion of apptime from '2000-01-01' to '2018-01-01'' at line 1
+delete from t for portion of othertime from '2000-01-01' to '2018-01-01';
+ERROR HY000: Period `othertime` is not found in table
+delete from t for portion of system_time from '2000-01-01' to '2018-01-01';
+ERROR 42000: You have an error in your SQL syntax; check the manual that corresponds to your MariaDB server version for the right syntax to use near 'of system_time from '2000-01-01' to '2018-01-01'' at line 1
+create or replace table t (id int, str text, s date, e date,
+period for apptime(s,e));
+insert into t values(1, 'data', '1999-01-01', '2018-12-12');
+insert into t values(1, 'other data', '1999-01-01', '2018-12-12');
+insert into t values(1, 'deleted', '2000-01-01', '2018-01-01');
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+show warnings;
+Level	Code	Message
+select * from t;
+id	str	s	e
+1	data	1999-01-01	2000-01-01
+1	data	2018-01-01	2018-12-12
+1	other data	1999-01-01	2000-01-01
+1	other data	2018-01-01	2018-12-12
+drop table t1;
+# SQL16, Part 2, 15.7 <Effect of deleting rows from base tables>,
+#        General rules, 8)b)i)
+# If the column descriptor that corresponds to the i-th field of BR
+# describes an identity column, a generated column, a system-time period
+# start column, or a system-time period end column, then let V i be
+# DEFAULT.
+# auto_increment field is updated
+create or replace table t (id int primary key auto_increment, s date, e date,
+period for apptime(s, e));
+insert into t values (default, '1999-01-01', '2018-12-12');
+select * from t;
+id	s	e
+1	1999-01-01	2018-12-12
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select * from t;
+id	s	e
+2	1999-01-01	2000-01-01
+3	2018-01-01	2018-12-12
+truncate t;
+# same for trigger case
+insert into t values (default, '1999-01-01', '2018-12-12');
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select * from t;
+id	s	e
+2	1999-01-01	2000-01-01
+3	2018-01-01	2018-12-12
+select * from log_tbl order by id;
+id	log
+1	>DEL: 1999-01-01, 2018-12-12
+2	>INS: 1999-01-01, 2000-01-01
+3	<INS: 1999-01-01, 2000-01-01
+4	>INS: 2018-01-01, 2018-12-12
+5	<INS: 2018-01-01, 2018-12-12
+6	<DEL: 1999-01-01, 2018-12-12
+# generated columns are updated
+create or replace table t (s date, e date,
+xs date as (s) stored, xe date as (e) stored,
+period for apptime(s, e));
+insert into t values('1999-01-01', '2018-12-12', default, default);
+select * from t;
+s	e	xs	xe
+1999-01-01	2018-12-12	1999-01-01	2018-12-12
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select * from t;
+s	e	xs	xe
+1999-01-01	2000-01-01	1999-01-01	2000-01-01
+2018-01-01	2018-12-12	2018-01-01	2018-12-12
+truncate t;
+# same for trigger case
+insert into t values('1999-01-01', '2018-12-12', default, default);
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select * from t;
+s	e	xs	xe
+1999-01-01	2000-01-01	1999-01-01	2000-01-01
+2018-01-01	2018-12-12	2018-01-01	2018-12-12
+select * from log_tbl order by id;
+id	log
+1	>DEL: 1999-01-01, 2018-12-12
+2	>INS: 1999-01-01, 2000-01-01
+3	<INS: 1999-01-01, 2000-01-01
+4	>INS: 2018-01-01, 2018-12-12
+5	<INS: 2018-01-01, 2018-12-12
+6	<DEL: 1999-01-01, 2018-12-12
+# View can't be used
+create or replace view v as select * from t;
+delete from v for portion of p from '2000-01-01' to '2018-01-01';
+ERROR 42S02: 'v' is a view
+# auto_increment field overflow
+create or replace table t (id tinyint auto_increment primary key,
+s date, e date, period for apptime(s,e));
+insert into t values(127, '1999-01-01', '2018-12-12');
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+ERROR 22003: Out of range value for column 'id' at row 1
+# same for trigger case
+# negotiate side effects of non-transactional MyISAM engine
+replace into t values(127, '1999-01-01', '2018-12-12');
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+ERROR 22003: Out of range value for column 'id' at row 1
+# custom constraint for period fields
+create or replace table t(id int, s date, e date, period for apptime(s,e),
+constraint dist2days check (datediff(e, s) >= 2));
+insert into t values(1, '1999-01-01', '2018-12-12'),
+(2, '1999-01-01', '1999-12-12');
+delete from t for portion of apptime from '1999-01-02' to '2018-12-12';
+ERROR 23000: CONSTRAINT `dist2days` failed for `test`.`t`
+# negotiate side effects of non-transactional MyISAM engine
+truncate t;
+insert into t values(1, '1999-01-01', '2018-12-12'),
+(2, '1999-01-01', '1999-12-12');
+delete from t for portion of apptime from '1999-01-01' to '2018-12-11';
+ERROR 23000: CONSTRAINT `dist2days` failed for `test`.`t`
+truncate t;
+insert into t values(1, '1999-01-01', '2018-12-12'),
+(2, '1999-01-01', '1999-12-12');
+delete from t for portion of apptime from '1999-01-03' to '2018-12-10';
+select *, datediff(e, s) from t;
+id	s	e	datediff(e, s)
+1	1999-01-01	1999-01-03	2
+1	2018-12-10	2018-12-12	2
+2	1999-01-01	1999-01-03	2
+# system_time columns are updated
+create or replace table t (
+s date, e date,
+row_start SYS_TYPE as row start invisible,
+row_end SYS_TYPE as row end invisible,
+period for apptime(s, e),
+period for system_time (row_start, row_end)) with system versioning;
+insert into t values('1999-01-01', '2018-12-12'),
+('1999-01-01', '1999-12-12');
+select row_start into @ins_time from t limit 1;
+select * from t order by s, e;
+s	e
+1999-01-01	1999-12-12
+1999-01-01	2018-12-12
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select *, if(row_start = @ins_time, "OLD", "NEW"), check_row(row_start, row_end)
+from t for system_time all
+order by s, e, row_start;
+s	e	if(row_start = @ins_time, "OLD", "NEW")	check_row(row_start, row_end)
+1999-01-01	1999-12-12	OLD	CURRENT ROW
+1999-01-01	2000-01-01	NEW	CURRENT ROW
+1999-01-01	2018-12-12	OLD	HISTORICAL ROW
+2018-01-01	2018-12-12	NEW	CURRENT ROW
+# same for trigger case
+delete from t;
+delete history from t;
+insert into t values('1999-01-01', '2018-12-12'),
+('1999-01-01', '1999-12-12');
+select row_start into @ins_time from t limit 1;
+select * from t order by s, e;
+s	e
+1999-01-01	1999-12-12
+1999-01-01	2018-12-12
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select *, if(row_start = @ins_time, "OLD", "NEW"), check_row(row_start, row_end)
+from t for system_time all
+order by s, e, row_start;
+s	e	if(row_start = @ins_time, "OLD", "NEW")	check_row(row_start, row_end)
+1999-01-01	1999-12-12	OLD	CURRENT ROW
+1999-01-01	2000-01-01	NEW	CURRENT ROW
+1999-01-01	2018-12-12	OLD	HISTORICAL ROW
+2018-01-01	2018-12-12	NEW	CURRENT ROW
+select * from log_tbl order by id;
+id	log
+1	>DEL: 1999-01-01, 2018-12-12
+2	>INS: 1999-01-01, 2000-01-01
+3	<INS: 1999-01-01, 2000-01-01
+4	>INS: 2018-01-01, 2018-12-12
+5	<INS: 2018-01-01, 2018-12-12
+6	<DEL: 1999-01-01, 2018-12-12
+create or replace database test;
diff --git a/mysql-test/suite/period/t/delete.test b/mysql-test/suite/period/t/delete.test
new file mode 100644
index 00000000000..689b06608fc
--- /dev/null
+++ b/mysql-test/suite/period/t/delete.test
@@ -0,0 +1,216 @@
+source suite/versioning/engines.inc;
+source suite/versioning/common.inc;
+
+create or replace table t (id int, s date, e date, period for apptime(s,e));
+
+insert into t values(1, '1999-01-01', '2018-12-12');
+insert into t values(1, '1999-01-01', '2017-01-01');
+insert into t values(1, '2017-01-01', '2019-01-01');
+insert into t values(2, '1998-01-01', '2018-12-12');
+insert into t values(3, '1997-01-01', '2015-01-01');
+insert into t values(4, '2016-01-01', '2020-01-01');
+insert into t values(5, '2010-01-01', '2015-01-01');
+
+create or replace table t1 (id int, s date, e date, period for apptime(s,e));
+insert t1 select * from t;
+create or replace table t2 (id int, s date, e date, period for apptime(s,e));
+insert t2 select * from t;
+create or replace table t3 (id int, s date, e date, period for apptime(s,e));
+insert t3 select * from t;
+
+--let $trig_cols=id, s, e
+--let $trig_table=t1
+--source suite/period/create_triggers.inc
+
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+delete from t1 for portion of APPTIME from '2000-01-01' to '2018-01-01';
+--sorted_result
+select * from t;
+--sorted_result
+select * from t1;
+select * from log_tbl order by id;
+
+--echo # INSERT trigger only also works
+--let $trig_cols=id, s, e
+--let $trig_table=t2
+--source suite/period/create_triggers.inc
+drop trigger tr1del_t2;
+drop trigger tr2del_t2;
+delete from t2 for portion of APPTIME from '2000-01-01' to '2018-01-01';
+select * from log_tbl order by id;
+
+--echo # removing BEFORE INSERT trigger enables internal substitution
+--echo # DELETE+INSERT -> UPDATE, but without any side effects.
+--echo # The optimization is disabled for non-transactional engines
+--let $trig_table=t3
+--source suite/period/create_triggers.inc
+drop trigger tr1ins_t3;
+delete from t3 for portion of APPTIME from '2000-01-01' to '2018-01-01';
+select * from log_tbl order by id;
+
+--echo # multi-table DELETE is not possible
+--error ER_PARSE_ERROR
+delete t, t1 from t1, t for portion of apptime from '2000-01-01' to '2018-01-01';
+
+--error ER_PARSE_ERROR
+delete t for portion of apptime from '2000-01-01' to '2018-01-01', t1 from t, t1;
+
+--echo # Here another check fails before parsing ends
+--error ER_UNKNOWN_TABLE
+delete t, t1 from t for portion of apptime from '2000-01-01' to '2018-01-01', t1;
+
+--error ER_PARSE_ERROR
+delete history from t2 for portion of apptime from '2000-01-01' to '2018-01-01';
+
+--error ER_PERIOD_NOT_FOUND
+delete from t for portion of othertime from '2000-01-01' to '2018-01-01';
+--error ER_PARSE_ERROR
+delete from t for portion of system_time from '2000-01-01' to '2018-01-01';
+
+create or replace table t (id int, str text, s date, e date,
+    period for apptime(s,e));
+
+insert into t values(1, 'data', '1999-01-01', '2018-12-12');
+insert into t values(1, 'other data', '1999-01-01', '2018-12-12');
+insert into t values(1, 'deleted', '2000-01-01', '2018-01-01');
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+show warnings;
+--sorted_result
+select * from t;
+
+drop table t1;
+
+--echo # SQL16, Part 2, 15.7 <Effect of deleting rows from base tables>,
+--echo #        General rules, 8)b)i)
+--echo # If the column descriptor that corresponds to the i-th field of BR
+--echo # describes an identity column, a generated column, a system-time period
+--echo # start column, or a system-time period end column, then let V i be
+--echo # DEFAULT.
+
+--echo # auto_increment field is updated
+create or replace table t (id int primary key auto_increment, s date, e date,
+                           period for apptime(s, e));
+insert into t values (default, '1999-01-01', '2018-12-12');
+select * from t;
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+--sorted_result
+select * from t;
+truncate t;
+--echo # same for trigger case
+insert into t values (default, '1999-01-01', '2018-12-12');
+--let $trig_table=t
+--source suite/period/create_triggers.inc
+
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+--sorted_result
+select * from t;
+select * from log_tbl order by id;
+
+--echo # generated columns are updated
+create or replace table t (s date, e date,
+                           xs date as (s) stored, xe date as (e) stored,
+                           period for apptime(s, e));
+insert into t values('1999-01-01', '2018-12-12', default, default);
+--sorted_result
+select * from t;
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+--sorted_result
+select * from t;
+truncate t;
+--echo # same for trigger case
+insert into t values('1999-01-01', '2018-12-12', default, default);
+--let $trig_table=t
+--source suite/period/create_triggers.inc
+
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+--sorted_result
+select * from t;
+select * from log_tbl order by id;
+
+--echo # View can't be used
+create or replace view v as select * from t;
+--error ER_IT_IS_A_VIEW
+delete from v for portion of p from '2000-01-01' to '2018-01-01';
+
+
+--echo # auto_increment field overflow
+create or replace table t (id tinyint auto_increment primary key,
+                           s date, e date, period for apptime(s,e));
+
+insert into t values(127, '1999-01-01', '2018-12-12');
+
+--error HA_ERR_AUTOINC_ERANGE
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+#select * from t;
+
+--echo # same for trigger case
+--let $trig_table=t
+--source suite/period/create_triggers.inc
+--echo # negotiate side effects of non-transactional MyISAM engine
+replace into t values(127, '1999-01-01', '2018-12-12');
+
+--error HA_ERR_AUTOINC_ERANGE
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+#select * from t;
+#select * from log_tbl order by id;
+
+--echo # custom constraint for period fields
+create or replace table t(id int, s date, e date, period for apptime(s,e),
+                          constraint dist2days check (datediff(e, s) >= 2));
+insert into t values(1, '1999-01-01', '2018-12-12'),
+                    (2, '1999-01-01', '1999-12-12');
+--error ER_CONSTRAINT_FAILED
+delete from t for portion of apptime from '1999-01-02' to '2018-12-12';
+--echo # negotiate side effects of non-transactional MyISAM engine
+truncate t;
+insert into t values(1, '1999-01-01', '2018-12-12'),
+                    (2, '1999-01-01', '1999-12-12');
+--error ER_CONSTRAINT_FAILED
+delete from t for portion of apptime from '1999-01-01' to '2018-12-11';
+truncate t;
+insert into t values(1, '1999-01-01', '2018-12-12'),
+                    (2, '1999-01-01', '1999-12-12');
+
+delete from t for portion of apptime from '1999-01-03' to '2018-12-10';
+--sorted_result
+select *, datediff(e, s) from t;
+
+
+--echo # system_time columns are updated
+--replace_result $sys_datatype_expl SYS_TYPE
+eval create or replace table t (
+    s date, e date,
+    row_start $sys_datatype_expl as row start invisible,
+    row_end $sys_datatype_expl as row end invisible,
+    period for apptime(s, e),
+    period for system_time (row_start, row_end)) with system versioning;
+insert into t values('1999-01-01', '2018-12-12'),
+                    ('1999-01-01', '1999-12-12');
+
+select row_start into @ins_time from t limit 1;
+select * from t order by s, e;
+
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select *, if(row_start = @ins_time, "OLD", "NEW"), check_row(row_start, row_end)
+       from t for system_time all
+       order by s, e, row_start;
+
+--echo # same for trigger case
+delete from t;
+delete history from t;
+insert into t values('1999-01-01', '2018-12-12'),
+                    ('1999-01-01', '1999-12-12');
+--let $trig_table=t
+--source suite/period/create_triggers.inc
+
+select row_start into @ins_time from t limit 1;
+select * from t order by s, e;
+
+delete from t for portion of apptime from '2000-01-01' to '2018-01-01';
+select *, if(row_start = @ins_time, "OLD", "NEW"), check_row(row_start, row_end)
+       from t for system_time all
+       order by s, e, row_start;
+select * from log_tbl order by id;
+
+
+create or replace database test;
diff --git a/mysql-test/suite/versioning/r/select.result b/mysql-test/suite/versioning/r/select.result
index be471ab8f30..f483bb49928 100644
--- a/mysql-test/suite/versioning/r/select.result
+++ b/mysql-test/suite/versioning/r/select.result
@@ -194,6 +194,40 @@ NULL	NULL	2	1
 NULL	NULL	3	1
 drop table t1;
 drop table t2;
+create or replace table t1(x int) with system versioning;
+insert into t1 values (1);
+delete from t1;
+insert into t1 values (2);
+delete from t1;
+insert into t1 values (3);
+delete from t1;
+select row_start into @start1 from t1 for system_time all where x = 1;
+select row_end into @end1 from t1 for system_time all where x = 1;
+select row_start into @start2 from t1 for system_time all where x = 2;
+select row_end into @end2 from t1 for system_time all where x = 2;
+select row_start into @start3 from t1 for system_time all where x = 3;
+select row_end into @end3 from t1 for system_time all where x = 3;
+select x as ASOF_x from t1 for system_time as of @start2;
+ASOF_x
+2
+select x as ASOF_x from t1 for system_time as of @end2;
+ASOF_x
+select x as FROMTO_x from t1 for system_time from @start1 to @end3;
+FROMTO_x
+1
+2
+3
+select x as FROMTO_x from t1 for system_time from @end1 to @start2;
+FROMTO_x
+select x as BETWAND_x from t1 for system_time between @start1 and @end3;
+BETWAND_x
+1
+2
+3
+select x as BETWAND_x from t1 for system_time between @end1 and @start2;
+BETWAND_x
+2
+drop table t1;
 create table t1(
 A int
 ) with system versioning;
diff --git a/mysql-test/suite/versioning/t/select.test b/mysql-test/suite/versioning/t/select.test
index b7c2d500da1..ebe4503401d 100644
--- a/mysql-test/suite/versioning/t/select.test
+++ b/mysql-test/suite/versioning/t/select.test
@@ -107,6 +107,32 @@ for system_time as of timestamp @t0 as t;
 drop table t1;
 drop table t2;
 
+# Query conditions check
+
+create or replace table t1(x int) with system versioning;
+insert into t1 values (1);
+delete from t1;
+insert into t1 values (2);
+delete from t1;
+insert into t1 values (3);
+delete from t1;
+
+select row_start into @start1 from t1 for system_time all where x = 1;
+select row_end into @end1 from t1 for system_time all where x = 1;
+select row_start into @start2 from t1 for system_time all where x = 2;
+select row_end into @end2 from t1 for system_time all where x = 2;
+select row_start into @start3 from t1 for system_time all where x = 3;
+select row_end into @end3 from t1 for system_time all where x = 3;
+
+select x as ASOF_x from t1 for system_time as of @start2;
+select x as ASOF_x from t1 for system_time as of @end2;
+select x as FROMTO_x from t1 for system_time from @start1 to @end3;
+select x as FROMTO_x from t1 for system_time from @end1 to @start2;
+select x as BETWAND_x from t1 for system_time between @start1 and @end3;
+select x as BETWAND_x from t1 for system_time between @end1 and @start2;
+
+drop table t1;
+
 # Wildcard expansion on hidden fields
 
 create table t1(
diff --git a/sql/lex.h b/sql/lex.h
index bcb085279b3..044e21a81f3 100644
--- a/sql/lex.h
+++ b/sql/lex.h
@@ -477,6 +477,7 @@ static SYMBOL symbols[] = {
   { "POINT",		SYM(POINT_SYM)},
   { "POLYGON",		SYM(POLYGON)},
   { "PORT",		SYM(PORT_SYM)},
+  { "PORTION",		SYM(PORTION_SYM)},
   { "PRECEDES",         SYM(PRECEDES_SYM)},
   { "PRECEDING",        SYM(PRECEDING_SYM)},
   { "PRECISION",	SYM(PRECISION)},
diff --git a/sql/share/errmsg-utf8.txt b/sql/share/errmsg-utf8.txt
index dcac7163ad6..b83ed4944a3 100644
--- a/sql/share/errmsg-utf8.txt
+++ b/sql/share/errmsg-utf8.txt
@@ -7944,3 +7944,5 @@ ER_MORE_THAN_ONE_PERIOD
         eng "Cannot specify more than one application-time period"
 ER_PERIOD_FIELD_WRONG_ATTRIBUTES
         eng "Period field %`s cannot be %s"
+ER_PERIOD_NOT_FOUND
+        eng "Period %`s is not found in table"
diff --git a/sql/sql_delete.cc b/sql/sql_delete.cc
index 1630d335559..a9109ce67c8 100644
--- a/sql/sql_delete.cc
+++ b/sql/sql_delete.cc
@@ -245,6 +245,50 @@ static bool record_should_be_deleted(THD *thd, TABLE *table, SQL_SELECT *sel,
   return false;
 }
 
+static
+int update_portion_of_time(THD *thd, TABLE *table,
+                           const vers_select_conds_t &period_conds,
+                           bool *inside_period)
+{
+  bool lcond= period_conds.field_start->val_datetime_packed(thd)
+              < period_conds.start.item->val_datetime_packed(thd);
+  bool rcond= period_conds.field_end->val_datetime_packed(thd)
+              > period_conds.end.item->val_datetime_packed(thd);
+
+  *inside_period= !lcond && !rcond;
+  if (*inside_period)
+    return 0;
+
+  DBUG_ASSERT(!table->triggers
+              || !table->triggers->has_triggers(TRG_EVENT_INSERT,
+                                                TRG_ACTION_BEFORE));
+
+  int res= 0;
+  Item *src= lcond ? period_conds.start.item : period_conds.end.item;
+  uint dst_fieldno= lcond ? table->s->period.end_fieldno
+                          : table->s->period.start_fieldno;
+
+  store_record(table, record[1]);
+  if (likely(!res))
+    res= src->save_in_field(table->field[dst_fieldno], true);
+
+  if (likely(!res))
+    res= table->update_generated_fields();
+
+  if(likely(!res))
+    res= table->file->ha_update_row(table->record[1], table->record[0]);
+
+  if (likely(!res) && table->triggers)
+    res= table->triggers->process_triggers(thd, TRG_EVENT_INSERT,
+                                           TRG_ACTION_AFTER, true);
+  restore_record(table, record[1]);
+
+  if (likely(!res) && lcond && rcond)
+    res= table->period_make_insert(period_conds.end.item,
+                                   table->field[table->s->period.start_fieldno]);
+
+  return res;
+}
 
 inline
 int TABLE::delete_row()
@@ -287,7 +331,7 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   bool		return_error= 0;
   ha_rows	deleted= 0;
   bool          reverse= FALSE;
-  bool          has_triggers;
+  bool          has_triggers= false;
   ORDER *order= (ORDER *) ((order_list && order_list->elements) ?
                            order_list->first : NULL);
   SELECT_LEX   *select_lex= thd->lex->first_select_lex();
@@ -298,7 +342,9 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   Explain_delete *explain;
   Delete_plan query_plan(thd->mem_root);
   Unique * deltempfile= NULL;
-  bool delete_record, delete_while_scanning;
+  bool delete_record= false;
+  bool delete_while_scanning;
+  bool portion_of_time_through_update;
   DBUG_ENTER("mysql_delete");
 
   query_plan.index= MAX_KEY;
@@ -313,6 +359,8 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   bool truncate_history= table_list->vers_conditions.is_set();
   if (truncate_history)
   {
+    DBUG_ASSERT(!table_list->period_conditions.is_set());
+
     if (table_list->is_view_or_derived())
     {
       my_error(ER_IT_IS_A_VIEW, MYF(0), table_list->table_name.str);
@@ -332,6 +380,18 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       table_list->on_expr= NULL;
     }
   }
+  if (table_list->has_period())
+  {
+    if (table_list->is_view_or_derived())
+    {
+      my_error(ER_IT_IS_A_VIEW, MYF(0), table_list->table_name.str);
+      DBUG_RETURN(true);
+    }
+
+    conds= select_lex->period_setup_conds(thd, table_list, conds);
+    if (!conds)
+      DBUG_RETURN(true);
+  }
 
   if (mysql_handle_list_of_derived(thd->lex, table_list, DT_MERGE_FOR_INSERT))
     DBUG_RETURN(TRUE);
@@ -423,12 +483,12 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
       - there should be no delete triggers associated with the table.
   */
 
-  has_triggers= (table->triggers &&
-                 table->triggers->has_delete_triggers());
+  has_triggers= table->triggers && table->triggers->has_delete_triggers();
+
   if (!with_select && !using_limit && const_cond_result &&
       (!thd->is_current_stmt_binlog_format_row() &&
        !has_triggers)
-      && !table->versioned(VERS_TIMESTAMP))
+      && !table->versioned(VERS_TIMESTAMP) && !table_list->has_period())
   {
     /* Update the table->file->stats.records number */
     table->file->info(HA_STATUS_VARIABLE | HA_STATUS_NO_LOCK);
@@ -598,7 +658,8 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   */
 
   if ((table->file->ha_table_flags() & HA_CAN_DIRECT_UPDATE_AND_DELETE) &&
-      !has_triggers && !binlog_is_row && !with_select)
+      !has_triggers && !binlog_is_row && !with_select &&
+      !table_list->has_period())
   {
     table->mark_columns_needed_for_delete();
     if (!table->check_virtual_columns_marked_for_read())
@@ -668,7 +729,10 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   if (unlikely(init_ftfuncs(thd, select_lex, 1)))
     goto got_error;
 
-  table->mark_columns_needed_for_delete();
+  if (table_list->has_period())
+    table->use_all_columns();
+  else
+    table->mark_columns_needed_for_delete();
 
   if ((table->file->ha_table_flags() & HA_CAN_FORCE_BULK_DELETE) &&
       !table->prepare_triggers_for_delete_stmt_or_event())
@@ -725,6 +789,24 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
     delete_record= true;
   }
 
+  /*
+    From SQL2016, Part 2, 15.7 <Effect of deleting rows from base table>,
+    General Rules, 8), we can conclude that DELETE FOR PORTTION OF time performs
+    0-2 INSERTS + DELETE. We can substitute INSERT+DELETE with one UPDATE, with
+    a condition of no side effects. The side effect is possible if there is a
+    BEFORE INSERT trigger, since it is the only one splitting DELETE and INSERT
+    operations.
+    Another possible side effect is related to tables of non-transactional
+    engines, since UPDATE is anyway atomic, and DELETE+INSERT is not.
+
+    This optimization is not possible for system-versioned table.
+  */
+  portion_of_time_through_update=
+          !(table->triggers && table->triggers->has_triggers(TRG_EVENT_INSERT,
+                                                             TRG_ACTION_BEFORE))
+          && !table->versioned()
+          && table->file->has_transactions();
+
   THD_STAGE_INFO(thd, stage_updating);
   while (likely(!(error=info.read_record())) && likely(!thd->killed) &&
          likely(!thd->is_error()))
@@ -748,7 +830,23 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
         break;
       }
 
-      error= table->delete_row();
+      if (table_list->has_period() && portion_of_time_through_update)
+      {
+        bool need_delete= true;
+        error= update_portion_of_time(thd, table, table_list->period_conditions,
+                                      &need_delete);
+        if (likely(!error) && need_delete)
+          error= table->delete_row();
+      }
+      else
+      {
+        error= table->delete_row();
+
+        if (likely(!error) && table_list->has_period()
+            && !portion_of_time_through_update)
+          error= table->insert_portion_of_time(thd, table_list->period_conditions);
+      }
+
       if (likely(!error))
       {
 	deleted++;
@@ -798,6 +896,8 @@ terminate_delete:
   }
   THD_STAGE_INFO(thd, stage_end);
   end_read_record(&info);
+  if (table_list->has_period())
+    table->file->ha_release_auto_increment();
   if (options & OPTION_QUICK)
     (void) table->file->extra(HA_EXTRA_NORMAL);
   ANALYZE_STOP_TRACKING(&explain->command_tracker);
@@ -967,7 +1067,13 @@ int mysql_prepare_delete(THD *thd, TABLE_LIST *table_list,
     DBUG_RETURN(TRUE);
   }
 
-  if (unique_table(thd, table_list, table_list->next_global, 0))
+  /*
+      Application-time periods: if FOR PORTION OF ... syntax used, DELETE
+      statement could issue delete_row's mixed with write_row's. This causes
+      problems for myisam and corrupts table, if deleting while scanning.
+   */
+  if (table_list->has_period()
+      || unique_table(thd, table_list, table_list->next_global, 0))
     *delete_while_scanning= false;
 
   if (select_lex->inner_refs_list.elements &&
diff --git a/sql/sql_lex.cc b/sql/sql_lex.cc
index d3f4793fb67..5b055c2899a 100644
--- a/sql/sql_lex.cc
+++ b/sql/sql_lex.cc
@@ -766,6 +766,7 @@ void LEX::start(THD *thd_arg)
   win_spec= NULL;
 
   vers_conditions.empty();
+  period_conditions.empty();
 
   is_lex_started= TRUE;
 
@@ -3606,6 +3607,19 @@ void LEX::set_trg_event_type_for_tables()
     break;
   }
 
+  if (period_conditions.is_set())
+  {
+    switch (sql_command)
+    {
+    case SQLCOM_DELETE:
+    case SQLCOM_UPDATE:
+    case SQLCOM_REPLACE:
+      new_trg_event_map |= trg2bit(TRG_EVENT_INSERT);
+    default:
+      break;
+    }
+  }
+
 
   /*
     Do not iterate over sub-selects, only the tables in the outermost
diff --git a/sql/sql_lex.h b/sql/sql_lex.h
index 43f20a4fa4c..0bc65382b27 100644
--- a/sql/sql_lex.h
+++ b/sql/sql_lex.h
@@ -1281,6 +1281,7 @@ public:
   /* push new Item_field into item_list */
   bool vers_push_field(THD *thd, TABLE_LIST *table, const LEX_CSTRING field_name);
 
+  Item* period_setup_conds(THD *thd, TABLE_LIST *table, Item *where);
   void init_query();
   void init_select();
   st_select_lex_unit* master_unit() { return (st_select_lex_unit*) master; }
@@ -3378,6 +3379,7 @@ public:
 
   /* System Versioning */
   vers_select_conds_t vers_conditions;
+  vers_select_conds_t period_conditions;
 
   inline void free_set_stmt_mem_root()
   {
diff --git a/sql/sql_select.cc b/sql/sql_select.cc
index de38cbf778d..ed891e83f8c 100644
--- a/sql/sql_select.cc
+++ b/sql/sql_select.cc
@@ -758,26 +758,163 @@ void vers_select_conds_t::print(String *str, enum_query_type query_type) const
   }
 }
 
-int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
+static
+Item* period_get_condition(THD *thd, TABLE_LIST *table, SELECT_LEX *select,
+                              vers_select_conds_t *conds, bool timestamp)
 {
-  DBUG_ENTER("SELECT_LEX::vers_setup_cond");
+  DBUG_ASSERT(table);
+  DBUG_ASSERT(table->table);
 #define newx new (thd->mem_root)
+  TABLE_SHARE *share= table->table->s;
+  const TABLE_SHARE::period_info_t *period= conds->period;
+
+  const LEX_CSTRING &fstart= period->start_field(share)->field_name;
+  const LEX_CSTRING &fend= period->end_field(share)->field_name;
+
+  conds->field_start= newx Item_field(thd, &select->context,
+                                      table->db.str, table->alias.str,
+                                      thd->make_clex_string(fstart));
+  conds->field_end=   newx Item_field(thd, &select->context,
+                                      table->db.str, table->alias.str,
+                                      thd->make_clex_string(fend));
+
+  Item *cond1= NULL, *cond2= NULL, *cond3= NULL, *curr= NULL;
+  if (timestamp)
+  {
+    MYSQL_TIME max_time;
+    switch (conds->type)
+    {
+    case SYSTEM_TIME_UNSPECIFIED:
+      thd->variables.time_zone->gmt_sec_to_TIME(&max_time, TIMESTAMP_MAX_VALUE);
+      max_time.second_part= TIME_MAX_SECOND_PART;
+      curr= newx Item_datetime_literal(thd, &max_time, TIME_SECOND_PART_DIGITS);
+      cond1= newx Item_func_eq(thd, conds->field_end, curr);
+      break;
+    case SYSTEM_TIME_AS_OF:
+      cond1= newx Item_func_le(thd, conds->field_start, conds->start.item);
+      cond2= newx Item_func_gt(thd, conds->field_end, conds->start.item);
+      break;
+    case SYSTEM_TIME_FROM_TO:
+      cond1= newx Item_func_lt(thd, conds->field_start, conds->end.item);
+      cond2= newx Item_func_gt(thd, conds->field_end, conds->start.item);
+      cond3= newx Item_func_lt(thd, conds->start.item, conds->end.item);
+      break;
+    case SYSTEM_TIME_BETWEEN:
+      cond1= newx Item_func_le(thd, conds->field_start, conds->end.item);
+      cond2= newx Item_func_gt(thd, conds->field_end, conds->start.item);
+      cond3= newx Item_func_le(thd, conds->start.item, conds->end.item);
+      break;
+    case SYSTEM_TIME_BEFORE:
+      cond1= newx Item_func_lt(thd, conds->field_end, conds->start.item);
+      break;
+    default:
+      DBUG_ASSERT(0);
+    }
+  }
+  else
+  {
+    DBUG_ASSERT(table->table->s && table->table->s->db_plugin);
 
-  TABLE_LIST *table;
+    Item *trx_id0= conds->start.item;
+    Item *trx_id1= conds->end.item;
+    if (conds->start.item && conds->start.unit == VERS_TIMESTAMP)
+    {
+      bool backwards= conds->type != SYSTEM_TIME_AS_OF;
+      trx_id0= newx Item_func_trt_id(thd, conds->start.item,
+                                     TR_table::FLD_TRX_ID, backwards);
+    }
+    if (conds->end.item && conds->end.unit == VERS_TIMESTAMP)
+    {
+      trx_id1= newx Item_func_trt_id(thd, conds->end.item,
+                                     TR_table::FLD_TRX_ID, false);
+    }
+
+    switch (conds->type)
+    {
+    case SYSTEM_TIME_UNSPECIFIED:
+      curr= newx Item_int(thd, ULONGLONG_MAX);
+      cond1= newx Item_func_eq(thd, conds->field_end, curr);
+      DBUG_ASSERT(!conds->start.item);
+      DBUG_ASSERT(!conds->end.item);
+      break;
+    case SYSTEM_TIME_AS_OF:
+      cond1= newx Item_func_trt_trx_sees_eq(thd, trx_id0, conds->field_start);
+      cond2= newx Item_func_trt_trx_sees(thd, conds->field_end, trx_id0);
+      DBUG_ASSERT(!conds->end.item);
+      break;
+    case SYSTEM_TIME_FROM_TO:
+      cond1= newx Item_func_trt_trx_sees(thd, trx_id1, conds->field_start);
+      cond2= newx Item_func_trt_trx_sees_eq(thd, conds->field_end, trx_id0);
+      cond3= newx Item_func_lt(thd, conds->start.item, conds->end.item);
+      break;
+    case SYSTEM_TIME_BETWEEN:
+      cond1= newx Item_func_trt_trx_sees_eq(thd, trx_id1, conds->field_start);
+      cond2= newx Item_func_trt_trx_sees_eq(thd, conds->field_end, trx_id0);
+      cond3= newx Item_func_le(thd, conds->start.item, conds->end.item);
+      break;
+    case SYSTEM_TIME_BEFORE:
+      cond1= newx Item_func_trt_trx_sees(thd, trx_id0, conds->field_end);
+      break;
+    default:
+      DBUG_ASSERT(0);
+    }
+  }
 
-  if (!thd->stmt_arena->is_conventional() &&
-      !thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
+  if (cond1)
   {
-    // statement is already prepared
-    DBUG_RETURN(0);
+    cond1= and_items(thd, cond2, cond1);
+    cond1= and_items(thd, cond3, cond1);
   }
+  return cond1;
+#undef newx
+}
 
-  if (thd->lex->is_view_context_analysis())
+static
+bool skip_setup_conds(THD *thd)
+{
+  return (!thd->stmt_arena->is_conventional()
+          && !thd->stmt_arena->is_stmt_prepare_or_first_sp_execute())
+         || thd->lex->is_view_context_analysis();
+}
+
+Item* SELECT_LEX::period_setup_conds(THD *thd, TABLE_LIST *tables, Item *where)
+{
+  DBUG_ENTER("SELECT_LEX::period_setup_conds");
+
+  if (skip_setup_conds(thd))
+    DBUG_RETURN(NULL);
+
+  DBUG_ASSERT(!tables->next_local && tables->table);
+
+  Item *result= NULL;
+  for (TABLE_LIST *table= tables; table; table= table->next_local)
+  {
+    if (!table->table)
+      continue;
+    vers_select_conds_t &conds= table->period_conditions;
+    if (!table->table->s->period.name.streq(conds.name))
+    {
+      my_error(ER_PERIOD_NOT_FOUND, MYF(0), conds.name.str);
+      DBUG_RETURN(NULL);
+    }
+
+    conds.period= &table->table->s->period;
+    result= and_items(thd, result,
+                      period_get_condition(thd, table, this, &conds, true));
+  }
+  DBUG_RETURN(and_items(thd, where, result));
+}
+
+int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
+{
+  DBUG_ENTER("SELECT_LEX::vers_setup_conds");
+
+  if (skip_setup_conds(thd))
     DBUG_RETURN(0);
 
   if (!versioned_tables)
   {
-    for (table= tables; table; table= table->next_local)
+    for (TABLE_LIST *table= tables; table; table= table->next_local)
     {
       if (table->table && table->table->versioned())
         versioned_tables++;
@@ -817,7 +954,7 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
     }
   }
 
-  for (table= tables; table; table= table->next_local)
+  for (TABLE_LIST *table= tables; table; table= table->next_local)
   {
     if (!table->table || !table->table->versioned())
       continue;
@@ -863,16 +1000,6 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
       lock_type= TL_READ; // ignore TL_WRITE, history is immutable anyway
     }
 
-    const LEX_CSTRING *fstart=
-        thd->make_clex_string(table->table->vers_start_field()->field_name);
-    const LEX_CSTRING *fend=
-        thd->make_clex_string(table->table->vers_end_field()->field_name);
-
-    Item *row_start=
-        newx Item_field(thd, &this->context, table->db.str, table->alias.str, fstart);
-    Item *row_end=
-        newx Item_field(thd, &this->context, table->db.str, table->alias.str, fend);
-
     bool timestamps_only= table->table->versioned(VERS_TIMESTAMP);
 
     if (vers_conditions.is_set())
@@ -892,101 +1019,16 @@ int SELECT_LEX::vers_setup_conds(THD *thd, TABLE_LIST *tables)
       }
     }
 
-    Item *cond1= NULL, *cond2= NULL, *cond3= NULL, *curr= NULL;
-    Item *point_in_time1= vers_conditions.start.item;
-    Item *point_in_time2= vers_conditions.end.item;
-    TABLE *t= table->table;
-    if (t->versioned(VERS_TIMESTAMP))
-    {
-      MYSQL_TIME max_time;
-      switch (vers_conditions.type)
-      {
-      case SYSTEM_TIME_UNSPECIFIED:
-        thd->variables.time_zone->gmt_sec_to_TIME(&max_time, TIMESTAMP_MAX_VALUE);
-        max_time.second_part= TIME_MAX_SECOND_PART;
-        curr= newx Item_datetime_literal(thd, &max_time, TIME_SECOND_PART_DIGITS);
-        cond1= newx Item_func_eq(thd, row_end, curr);
-        break;
-      case SYSTEM_TIME_AS_OF:
-        cond1= newx Item_func_le(thd, row_start, point_in_time1);
-        cond2= newx Item_func_gt(thd, row_end, point_in_time1);
-        break;
-      case SYSTEM_TIME_FROM_TO:
-        cond1= newx Item_func_lt(thd, row_start, point_in_time2);
-        cond2= newx Item_func_gt(thd, row_end, point_in_time1);
-        cond3= newx Item_func_lt(thd, point_in_time1, point_in_time2);
-        break;
-      case SYSTEM_TIME_BETWEEN:
-        cond1= newx Item_func_le(thd, row_start, point_in_time2);
-        cond2= newx Item_func_gt(thd, row_end, point_in_time1);
-        cond3= newx Item_func_le(thd, point_in_time1, point_in_time2);
-        break;
-      case SYSTEM_TIME_BEFORE:
-        cond1= newx Item_func_lt(thd, row_end, point_in_time1);
-        break;
-      default:
-        DBUG_ASSERT(0);
-      }
-    }
-    else
-    {
-      DBUG_ASSERT(table->table->s && table->table->s->db_plugin);
-
-      Item *trx_id0, *trx_id1;
-
-      switch (vers_conditions.type)
-      {
-      case SYSTEM_TIME_UNSPECIFIED:
-        curr= newx Item_int(thd, ULONGLONG_MAX);
-        cond1= newx Item_func_eq(thd, row_end, curr);
-        break;
-      case SYSTEM_TIME_AS_OF:
-        trx_id0= vers_conditions.start.unit == VERS_TIMESTAMP
-          ? newx Item_func_trt_id(thd, point_in_time1, TR_table::FLD_TRX_ID)
-          : point_in_time1;
-        cond1= newx Item_func_trt_trx_sees_eq(thd, trx_id0, row_start);
-        cond2= newx Item_func_trt_trx_sees(thd, row_end, trx_id0);
-        break;
-      case SYSTEM_TIME_FROM_TO:
-	cond3= newx Item_func_lt(thd, point_in_time1, point_in_time2);
-        /* fall through */
-      case SYSTEM_TIME_BETWEEN:
-        trx_id0= vers_conditions.start.unit == VERS_TIMESTAMP
-          ? newx Item_func_trt_id(thd, point_in_time1, TR_table::FLD_TRX_ID, true)
-          : point_in_time1;
-        trx_id1= vers_conditions.end.unit == VERS_TIMESTAMP
-          ? newx Item_func_trt_id(thd, point_in_time2, TR_table::FLD_TRX_ID, false)
-          : point_in_time2;
-        cond1= vers_conditions.type == SYSTEM_TIME_FROM_TO
-          ? newx Item_func_trt_trx_sees(thd, trx_id1, row_start)
-          : newx Item_func_trt_trx_sees_eq(thd, trx_id1, row_start);
-        cond2= newx Item_func_trt_trx_sees_eq(thd, row_end, trx_id0);
-	if (!cond3)
-	  cond3= newx Item_func_le(thd, point_in_time1, point_in_time2);
-        break;
-      case SYSTEM_TIME_BEFORE:
-        trx_id0= vers_conditions.start.unit == VERS_TIMESTAMP
-          ? newx Item_func_trt_id(thd, point_in_time1, TR_table::FLD_TRX_ID, true)
-          : point_in_time1;
-        cond1= newx Item_func_trt_trx_sees(thd, trx_id0, row_end);
-        break;
-      default:
-        DBUG_ASSERT(0);
-      }
-    }
-
-    if (cond1)
-    {
-      cond1= and_items(thd, cond2, cond1);
-      cond1= and_items(thd, cond3, cond1);
-      table->on_expr= and_items(thd, table->on_expr, cond1);
-    }
-
+    vers_conditions.period = &table->table->s->vers;
+    Item *cond= period_get_condition(thd, table, this, &vers_conditions,
+                                     timestamps_only);
+    if (cond)
+      table->on_expr= and_items(thd, table->on_expr, cond);
     table->vers_conditions.type= SYSTEM_TIME_ALL;
+
   } // for (table= tables; ...)
 
   DBUG_RETURN(0);
-#undef newx
 }
 
 /*****************************************************************************
diff --git a/sql/sql_yacc.yy b/sql/sql_yacc.yy
index 1f0bf717ed4..85a0b14f96c 100644
--- a/sql/sql_yacc.yy
+++ b/sql/sql_yacc.yy
@@ -1024,6 +1024,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  PERCENT_RANK_SYM
 %token  PERCENTILE_CONT_SYM
 %token  PERCENTILE_DISC_SYM
+%token  PORTION_SYM                   /* SQL-2016-R */
 %token  POSITION_SYM                  /* SQL-2003-N */
 %token  PRECISION                     /* SQL-2003-R */
 %token  PRIMARY_SYM                   /* SQL-2003-R */
@@ -1812,7 +1813,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         opt_default_time_precision
         case_stmt_body opt_bin_mod opt_for_system_time_clause
         opt_if_exists_table_element opt_if_not_exists_table_element
-        opt_recursive opt_format_xid
+        opt_recursive opt_format_xid opt_for_portion_of_time_clause
 
 %type <object_ddl_options>
         create_or_replace
@@ -9443,6 +9444,26 @@ history_point:
             $$= Vers_history_point($1, $2);
           }
         ;
+opt_for_portion_of_time_clause:
+          /* empty */
+          {
+            $$= false;
+          }
+        | FOR_SYM PORTION_SYM OF_SYM remember_tok_start ident FROM
+          bit_expr TO_SYM bit_expr
+          {
+            if (unlikely(0 == strcasecmp($5.str, "SYSTEM_TIME")))
+            {
+              thd->parse_error(ER_SYNTAX_ERROR, $4);
+              MYSQL_YYABORT;
+            }
+            Lex->period_conditions.init(SYSTEM_TIME_FROM_TO,
+                                        Vers_history_point(VERS_TIMESTAMP, $7),
+                                        Vers_history_point(VERS_TIMESTAMP, $9),
+                                        $5);
+            $$= true;
+          }
+        ;
 
 opt_for_system_time_clause:
           /* empty */
@@ -13686,8 +13707,16 @@ delete_single_table:
           }
         ;
 
+delete_single_table_for_period:
+        delete_single_table opt_for_portion_of_time_clause
+        {
+          if ($2)
+            Lex->last_table()->period_conditions= Lex->period_conditions;
+        }
+      ;
+
 single_multi:
-          delete_single_table
+          delete_single_table_for_period
           opt_where_clause
           opt_order_clause
           delete_limit_clause
diff --git a/sql/sql_yacc_ora.yy b/sql/sql_yacc_ora.yy
index 46f4ee8f7a5..4d46552d8a2 100644
--- a/sql/sql_yacc_ora.yy
+++ b/sql/sql_yacc_ora.yy
@@ -519,6 +519,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
 %token  PERCENT_RANK_SYM
 %token  PERCENTILE_CONT_SYM
 %token  PERCENTILE_DISC_SYM
+%token  PORTION_SYM                   /* SQL-2016-R */
 %token  POSITION_SYM                  /* SQL-2003-N */
 %token  PRECISION                     /* SQL-2003-R */
 %token  PRIMARY_SYM                   /* SQL-2003-R */
@@ -1314,7 +1315,7 @@ bool my_yyoverflow(short **a, YYSTYPE **b, size_t *yystacksize);
         opt_default_time_precision
         case_stmt_body opt_bin_mod opt_for_system_time_clause
         opt_if_exists_table_element opt_if_not_exists_table_element
-        opt_recursive opt_format_xid
+        opt_recursive opt_format_xid opt_for_portion_of_time_clause
 
 %type <object_ddl_options>
         create_or_replace
@@ -9467,6 +9468,23 @@ history_point:
           }
         ;
 
+opt_for_portion_of_time_clause:
+          /* empty */
+          {
+            $$= false;
+          }
+        | FOR_SYM PORTION_SYM OF_SYM ident FROM history_point TO_SYM history_point
+          {
+            if (unlikely(0 == strcasecmp($4.str, "SYSTEM_TIME")))
+            {
+              thd->parse_error(ER_SYNTAX_ERROR, $4.str);
+              MYSQL_YYABORT;
+            }
+            $$= true;
+            Lex->period_conditions.init(SYSTEM_TIME_FROM_TO, $6, $8, $4);
+          }
+        ;
+
 opt_for_system_time_clause:
           /* empty */
           {
@@ -13730,11 +13748,20 @@ delete_single_table:
               MYSQL_YYABORT;
             YYPS->m_lock_type= TL_READ_DEFAULT;
             YYPS->m_mdl_type= MDL_SHARED_READ;
+            Lex->last_table()->period_conditions= Lex->period_conditions;
+          }
+        ;
+
+delete_single_table_for_period:
+          delete_single_table opt_for_portion_of_time_clause
+          {
+            Lex->last_table()->period_conditions= Lex->period_conditions;
           }
         ;
 
+
 single_multi:
-          delete_single_table
+          delete_single_table_for_period
           opt_where_clause
           opt_order_clause
           delete_limit_clause
diff --git a/sql/table.cc b/sql/table.cc
index c31b93d4d8e..863d303efcc 100644
--- a/sql/table.cc
+++ b/sql/table.cc
@@ -6682,9 +6682,9 @@ void TABLE::mark_columns_needed_for_delete()
 
   if (s->versioned)
   {
-    bitmap_set_bit(read_set, s->vers.start_field(s)->field_index);
-    bitmap_set_bit(read_set, s->vers.end_field(s)->field_index);
-    bitmap_set_bit(write_set, s->vers.end_field(s)->field_index);
+    bitmap_set_bit(read_set, s->vers.start_fieldno);
+    bitmap_set_bit(read_set, s->vers.end_fieldno);
+    bitmap_set_bit(write_set, s->vers.end_fieldno);
   }
 }
 
@@ -8091,6 +8091,69 @@ int TABLE::update_default_fields(bool update_command, bool ignore_errors)
   DBUG_RETURN(res);
 }
 
+int TABLE::update_generated_fields()
+{
+  int res= 0;
+  if (found_next_number_field)
+  {
+    next_number_field= found_next_number_field;
+    res= found_next_number_field->set_default();
+    if (likely(!res))
+      res= file->update_auto_increment();
+  }
+
+  if (likely(!res) && vfield)
+    res= update_virtual_fields(file, VCOL_UPDATE_FOR_WRITE);
+  if (likely(!res) && versioned())
+    vers_update_fields();
+  if (likely(!res))
+    res= verify_constraints(false) == VIEW_CHECK_ERROR;
+  return res;
+}
+
+int TABLE::period_make_insert(Item *src, Field *dst)
+{
+  THD *thd= in_use;
+
+  store_record(this, record[1]);
+  int res= src->save_in_field(dst, true);
+
+  if (likely(!res))
+    res= update_generated_fields();
+
+  if (likely(!res) && triggers)
+    res= triggers->process_triggers(thd, TRG_EVENT_INSERT,
+                                    TRG_ACTION_BEFORE, true);
+
+  if (likely(!res))
+    res = file->ha_write_row(record[0]);
+
+  if (likely(!res) && triggers)
+    res= triggers->process_triggers(thd, TRG_EVENT_INSERT,
+                                    TRG_ACTION_AFTER, true);
+
+  restore_record(this, record[1]);
+  return res;
+}
+
+int TABLE::insert_portion_of_time(THD *thd,
+                                  const vers_select_conds_t &period_conds)
+{
+  bool lcond= period_conds.field_start->val_datetime_packed(thd)
+              < period_conds.start.item->val_datetime_packed(thd);
+  bool rcond= period_conds.field_end->val_datetime_packed(thd)
+              > period_conds.end.item->val_datetime_packed(thd);
+
+  int res= 0;
+  if (lcond)
+    res= period_make_insert(period_conds.start.item,
+                            field[s->period.end_fieldno]);
+  if (likely(!res) && rcond)
+    res= period_make_insert(period_conds.end.item,
+                            field[s->period.start_fieldno]);
+
+  return res;
+}
 
 void TABLE::vers_update_fields()
 {
diff --git a/sql/table.h b/sql/table.h
index 0dc4728d0ce..fb77d6b33c9 100644
--- a/sql/table.h
+++ b/sql/table.h
@@ -1091,6 +1091,8 @@ typedef Bitmap<MAX_FIELDS> Field_map;
 
 class SplM_opt_info;
 
+struct vers_select_conds_t;
+
 struct TABLE
 {
   TABLE() {}                               /* Remove gcc warning */
@@ -1574,6 +1576,9 @@ public:
   ulonglong vers_start_id() const;
   ulonglong vers_end_id() const;
 
+  int update_generated_fields();
+  int period_make_insert(Item *src, Field *dst);
+  int insert_portion_of_time(THD *thd, const vers_select_conds_t &period_conds);
   int delete_row();
   void vers_update_fields();
   void vers_update_end();
@@ -1886,6 +1891,10 @@ struct vers_select_conds_t
   bool used:1;
   Vers_history_point start;
   Vers_history_point end;
+  Lex_ident name;
+
+  Item_field *field_start;
+  Item_field *field_end;
 
   const TABLE_SHARE::period_info_t *period;
 
@@ -1899,12 +1908,14 @@ struct vers_select_conds_t
 
   void init(vers_system_time_t _type,
             Vers_history_point _start= Vers_history_point(),
-            Vers_history_point _end= Vers_history_point())
+            Vers_history_point _end= Vers_history_point(),
+            Lex_ident          _name= "SYSTEM_TIME")
   {
     type= _type;
     used= false;
     start= _start;
     end= _end;
+    name= _name;
   }
 
   void print(String *str, enum_query_type query_type) const;
@@ -2002,6 +2013,7 @@ struct TABLE_LIST
     init_one_table(&table_arg->s->db, &table_arg->s->table_name,
                    NULL, lock_type);
     table= table_arg;
+    vers_conditions.name= table->s->vers.name;
   }
 
   inline void init_one_table_for_prelocking(const LEX_CSTRING *db_arg,
@@ -2444,6 +2456,12 @@ struct TABLE_LIST
 
   /* System Versioning */
   vers_select_conds_t vers_conditions;
+  vers_select_conds_t period_conditions;
+
+  bool has_period() const
+  {
+    return period_conditions.is_set();
+  }
 
   /**
      @brief