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

mysql-test/suite/period/create_triggers.inc

+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;

mysql-test/suite/period/t/delete.test

+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;

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;

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(

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)},

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"

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 @@ bool mysql_delete(THD *thd, TABLE_LIST *table_list, COND *conds,
   }
   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 &&

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

sql/sql_lex.h

@@ -1281,6 +1281,7 @@ class st_select_lex: public st_select_lex_node
   /* 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 @@ struct LEX: public Query_tables_list
 
   /* System Versioning */
   vers_select_conds_t vers_conditions;
+  vers_select_conds_t period_conditions;
 
   inline void free_set_stmt_mem_root()
   {

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

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

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()
 {

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 @@ struct TABLE
   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