MDEV-15364 FOREIGN CASCADE operations in system versioned referenced tables

Merge pull request #667

mysql-test/suite/versioning/common.inc

@@ -77,4 +77,26 @@ if ($MTR_COMBINATION_TRX_ID)
   let $sys_datatype_expl_uc= BIGINT(20) UNSIGNED;
   let $sys_datatype_max= 18446744073709551615;
 }
+
+eval create or replace function current_row(sys_trx_end $sys_datatype_expl)
+returns int
+deterministic
+  return sys_trx_end = $sys_datatype_max;
+
+delimiter ~~;
+eval create or replace function check_row(row_start $sys_datatype_expl, row_end $sys_datatype_expl)
+returns varchar(255)
+deterministic
+begin
+  if row_end < row_start then
+    return "ERROR: row_end < row_start";
+  elseif row_end = row_start then
+    return "ERROR: row_end == row_start";
+  elseif current_row(row_end) then
+    return "CURRENT ROW";
+  end if;
+  return "HISTORICAL ROW";
+end~~
+delimiter ;~~
+
 --enable_query_log

mysql-test/suite/versioning/common_finish.inc

@@ -4,4 +4,6 @@ drop procedure verify_vtq_dummy;
 drop function sys_commit_ts;
 drop procedure concat_exec2;
 drop procedure concat_exec3;
+drop function current_row;
+drop function check_row;
 --enable_query_log

mysql-test/suite/versioning/r/foreign.result

@@ -6,8 +6,8 @@ id int unique key
 ) engine innodb;
 create table child(
 parent_id int,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end),
 foreign key(parent_id) references parent(id)
 on delete restrict
@@ -25,7 +25,7 @@ update parent set id=id+1;
 ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails (`test`.`child`, CONSTRAINT `child_ibfk_1` FOREIGN KEY (`parent_id`) REFERENCES `parent` (`id`))
 delete from child;
 update parent set id=id+1;
-select * from child for system_time from timestamp 0 to timestamp now(6);
+select * from child for system_time all;
 parent_id
 1
 1
@@ -39,8 +39,8 @@ id int(10) unsigned unique key
 ) engine innodb;
 create table child(
 parent_id int(10) unsigned primary key,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end),
 foreign key(parent_id) references parent(id)
 ) engine innodb with system versioning;
@@ -58,19 +58,38 @@ id int unique key
 ) engine innodb;
 create table child(
 parent_id int,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end),
 foreign key(parent_id) references parent(id)
 on delete cascade
 on update cascade
 ) engine innodb with system versioning;
-ERROR HY000: CASCADE is not supported for TIMESTAMP(6) AS ROW START/END system-versioned tables
+insert into parent values(1);
+insert into child values(1);
+delete from parent where id = 1;
+select * from child;
+parent_id
+select * from child for system_time all;
+parent_id
+1
+insert into parent values(1);
+insert into child values(1);
+update parent set id = id + 1;
+select * from child;
+parent_id
+2
+select * from child for system_time all;
+parent_id
+1
+1
+2
+drop table child;
 drop table parent;
 create or replace table parent (
 id int primary key,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end)
 ) with system versioning
 engine innodb;
@@ -97,8 +116,8 @@ engine innodb;
 create or replace table child (
 id int primary key,
 parent_id int not null,
-row_start timestamp(6) as row start invisible,
+row_start SYS_DATATYPE as row start invisible,
-row_end timestamp(6) as row end invisible,
+row_end SYS_DATATYPE as row end invisible,
 period for system_time(row_start, row_end),
 constraint `parent-fk`
   foreign key (parent_id) references parent (id)
@@ -106,32 +125,67 @@ on delete cascade
 on update restrict
 ) with system versioning
 engine innodb;
-ERROR HY000: CASCADE is not supported for TIMESTAMP(6) AS ROW START/END system-versioned tables
+insert into parent (id) values (3);
+insert into child (id, parent_id) values (3, 3);
+delete from parent;
+select * from child;
+id	parent_id
+select *, check_row(row_start, row_end) from child for system_time all;
+id	parent_id	check_row(row_start, row_end)
+3	3	HISTORICAL ROW
+drop table child;
 drop table parent;
 #################
 # Test SET NULL #
 #################
-create table parent(
+create or replace table parent(
 id int unique key
 ) engine innodb;
-create table child(
+create or replace table child(
 parent_id int,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end),
 foreign key(parent_id) references parent(id)
 on delete set null
 on update set null
 ) engine innodb with system versioning;
-ERROR HY000: SET NULL is not supported for TIMESTAMP(6) AS ROW START/END system-versioned tables
+insert into parent values(1);
+insert into child values(1);
+delete from child;
+insert into child values(1);
+delete from parent where id = 1;
+select * from child;
+parent_id
+NULL
+select *, current_row(sys_end) as current_row from child for system_time all order by sys_end;
+parent_id	current_row
+1	0
+1	0
+NULL	1
+delete from child;
+insert into parent values(1);
+insert into child values(1);
+update parent set id= id + 1;
+select * from child;
+parent_id
+NULL
+select *, current_row(sys_end) as current_row from child for system_time all order by sys_end;
+parent_id	current_row
+1	0
+1	0
+NULL	0
+1	0
+NULL	1
+drop table child;
 drop table parent;
 ###########################
 # Parent table is foreign #
 ###########################
 create or replace table parent(
 id int unique key,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end)
 ) engine innodb with system versioning;
 create or replace table child(
@@ -162,16 +216,16 @@ drop table parent;
 create or replace table a (
 cola int(10) primary key,
 v_cola int(10) as (cola mod 10) virtual,
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end)
 ) engine=innodb with system versioning;
 create index v_cola on a (v_cola);
 create or replace table b(
 cola int(10),
 v_cola int(10),
-sys_start timestamp(6) as row start invisible,
+sys_start SYS_DATATYPE as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end)
 ) engine=innodb with system versioning;
 alter table b add constraint `v_cola_fk`
@@ -181,3 +235,58 @@ insert into b(cola, v_cola) values (10,2);
 delete from a;
 ERROR 23000: Cannot delete or update a parent row: a foreign key constraint fails (`test`.`b`, CONSTRAINT `v_cola_fk` FOREIGN KEY (`v_cola`) REFERENCES `a` (`v_cola`))
 drop table b, a;
+###############################################
+# CASCADE UPDATE foreign not system versioned #
+###############################################
+create or replace table parent (
+id smallint unsigned not null auto_increment,
+value int unsigned not null,
+primary key (id, value)
+) engine = innodb;
+create or replace table child (
+id mediumint unsigned not null auto_increment primary key,
+parent_id smallint unsigned not null,
+parent_value int unsigned not null,
+sys_start SYS_DATATYPE as row start invisible,
+sys_end SYS_DATATYPE as row end invisible,
+period for system_time(sys_start, sys_end),
+constraint `fk_child_parent`
+    foreign key (parent_id, parent_value) references parent (id, value)
+on delete cascade
+on update cascade
+) engine = innodb with system versioning;
+create or replace table subchild (
+id int not null auto_increment primary key,
+parent_id smallint unsigned not null,
+parent_value int unsigned not null,
+constraint `fk_subchild_child_parent`
+    foreign key (parent_id, parent_value) references child (parent_id, parent_value)
+on delete cascade
+on update cascade
+) engine=innodb;
+insert into parent (value) values (23);
+select id, value from parent into @id, @value;
+insert into child values (default, @id, @value);
+insert into subchild values (default, @id, @value);
+select parent_id from subchild;
+parent_id
+1
+update parent set id = 11, value = value + 1;
+select parent_id from subchild;
+parent_id
+11
+select * from child;
+id	parent_id	parent_value
+1	11	24
+delete from parent;
+select count(*) from child;
+count(*)
+0
+select * from child for system_time all;
+id	parent_id	parent_value
+1	1	23
+1	11	24
+select count(*) from subchild;
+count(*)
+0
+drop table subchild, child, parent;

mysql-test/suite/versioning/t/foreign.test

@@ -8,6 +8,7 @@ create table parent(
   id int unique key
 ) engine innodb;
 
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create table child(
   parent_id int,
   sys_start $sys_datatype_expl as row start invisible,
@@ -32,7 +33,7 @@ insert into child values(1);
 update parent set id=id+1;
 delete from child;
 update parent set id=id+1;
-select * from child for system_time from timestamp 0 to timestamp now(6);
+select * from child for system_time all;
 
 drop table child;
 drop table parent;
@@ -45,6 +46,7 @@ create table parent(
   id int(10) unsigned unique key
 ) engine innodb;
 
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create table child(
   parent_id int(10) unsigned primary key,
   sys_start $sys_datatype_expl as row start invisible,
@@ -70,7 +72,7 @@ create table parent(
   id int unique key
 ) engine innodb;
 
---disable_abort_on_error
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create table child(
   parent_id int,
   sys_start $sys_datatype_expl as row start invisible,
@@ -80,14 +82,10 @@ eval create table child(
     on delete cascade
     on update cascade
 ) engine innodb with system versioning;
---enable_abort_on_error
 
-if ($MTR_COMBINATION_TRX_ID) {
 insert into parent values(1);
 insert into child values(1);
 
-delete from parent where id = 1;
-delete from child where parent_id = 1;
 delete from parent where id = 1;
 select * from child;
 select * from child for system_time all;
@@ -99,8 +97,9 @@ select * from child;
 select * from child for system_time all;
 
 drop table child;
-}
 drop table parent;
+
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create or replace table parent (
   id int primary key,
   sys_start $sys_datatype_expl as row start invisible,
@@ -132,7 +131,7 @@ create or replace table parent (
 )
 engine innodb;
 
---disable_abort_on_error
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create or replace table child (
   id int primary key,
   parent_id int not null,
@@ -145,33 +144,26 @@ eval create or replace table child (
       on update restrict
 ) with system versioning
 engine innodb;
---enable_abort_on_error
 
-if ($MTR_COMBINATION_TRX_ID) {
 insert into parent (id) values (3);
 insert into child (id, parent_id) values (3, 3);
---echo ## FIXME: #415 update of foreign constraints is disabled
-delete from child;
---echo ## FIXME END
 delete from parent;
 select * from child;
---replace_result $sys_datatype_max MAXVAL
+select *, check_row(row_start, row_end) from child for system_time all;
-eval select *, row_start < row_end, row_end < $sys_datatype_max from child for system_time all;
 
 drop table child;
-}
 drop table parent;
 
 --echo #################
 --echo # Test SET NULL #
 --echo #################
 
-create table parent(
+create or replace table parent(
   id int unique key
 ) engine innodb;
 
---disable_abort_on_error
+--replace_result $sys_datatype_expl SYS_DATATYPE
-eval create table child(
+eval create or replace table child(
   parent_id int,
   sys_start $sys_datatype_expl as row start invisible,
   sys_end $sys_datatype_expl as row end invisible,
@@ -180,41 +172,31 @@ eval create table child(
     on delete set null
     on update set null
 ) engine innodb with system versioning;
---enable_abort_on_error
 
-if ($MTR_COMBINATION_TRX_ID) {
 insert into parent values(1);
 insert into child values(1);
 delete from child;
 insert into child values(1);
 
---echo ## FIXME: #415 update of foreign constraints is disabled
-delete from child where parent_id = 1;
---echo ## FIXME END
 delete from parent where id = 1;
 select * from child;
-select * from child for system_time from timestamp 0 to timestamp now(6);
+select *, current_row(sys_end) as current_row from child for system_time all order by sys_end;
 delete from child;
 
 insert into parent values(1);
 insert into child values(1);
-## FIXME: #415 update of foreign constraints is disabled
+update parent set id= id + 1;
-if (0)
-{
-update parent set id=id+1;
 select * from child;
-select * from child for system_time from timestamp 0 to timestamp now(6);
+select *, current_row(sys_end) as current_row from child for system_time all order by sys_end;
-}
-## FIXME END
 
 drop table child;
-}
 drop table parent;
 
 --echo ###########################
 --echo # Parent table is foreign #
 --echo ###########################
 
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create or replace table parent(
   id int unique key,
   sys_start $sys_datatype_expl as row start invisible,
@@ -254,6 +236,7 @@ drop table parent;
 --echo # crash on DELETE #
 --echo ###################
 
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create or replace table a (
   cola int(10) primary key,
   v_cola int(10) as (cola mod 10) virtual,
@@ -264,6 +247,7 @@ eval create or replace table a (
 
 create index v_cola on a (v_cola);
 
+--replace_result $sys_datatype_expl SYS_DATATYPE
 eval create or replace table b(
   cola int(10),
   v_cola int(10),
@@ -282,4 +266,55 @@ delete from a;
 
 drop table b, a;
 
+--echo ###############################################
+--echo # CASCADE UPDATE foreign not system versioned #
+--echo ###############################################
+create or replace table parent (
+	id smallint unsigned not null auto_increment,
+	value int unsigned not null,
+	primary key (id, value)
+) engine = innodb;
+
+--replace_result $sys_datatype_expl SYS_DATATYPE
+eval create or replace table child (
+  id mediumint unsigned not null auto_increment primary key,
+  parent_id smallint unsigned not null,
+  parent_value int unsigned not null,
+  sys_start $sys_datatype_expl as row start invisible,
+  sys_end $sys_datatype_expl as row end invisible,
+  period for system_time(sys_start, sys_end),
+  constraint `fk_child_parent`
+    foreign key (parent_id, parent_value) references parent (id, value)
+    on delete cascade
+    on update cascade
+) engine = innodb with system versioning;
+
+create or replace table subchild (
+  id int not null auto_increment primary key,
+  parent_id smallint unsigned not null,
+  parent_value int unsigned not null,
+  constraint `fk_subchild_child_parent`
+    foreign key (parent_id, parent_value) references child (parent_id, parent_value)
+    on delete cascade
+    on update cascade
+) engine=innodb;
+
+insert into parent (value) values (23);
+select id, value from parent into @id, @value;
+insert into child values (default, @id, @value);
+insert into subchild values (default, @id, @value);
+
+select parent_id from subchild;
+update parent set id = 11, value = value + 1;
+select parent_id from subchild;
+select * from child;
+
+delete from parent;
+select count(*) from child;
+select * from child for system_time all;
+select count(*) from subchild;
+
+drop table subchild, child, parent;
+
+
 --source suite/versioning/common_finish.inc

sql/sql_class.cc

@@ -5177,6 +5177,20 @@ extern "C" bool thd_is_strict_mode(const MYSQL_THD thd)
 }
 
 
+/**
+  Get query start time as SQL field data.
+  Needed by InnoDB.
+  @param thd	Thread object
+  @param buf	Buffer to hold start time data
+*/
+void thd_get_query_start_data(THD *thd, char *buf)
+{
+  LEX_CSTRING field_name;
+  Field_timestampf f((uchar *)buf, NULL, 0, Field::NONE, &field_name, NULL, 6);
+  f.store_TIME(thd->query_start(), thd->query_start_sec_part());
+}
+
+
 /*
   Interface for MySQL Server, plugins and storage engines to report
   when they are going to sleep/stall.

sql/sql_table.cc

@@ -4479,19 +4479,6 @@ static bool vers_prepare_keys(THD *thd, HA_CREATE_INFO *create_info,
   Key *key= NULL;
   while ((key=key_it++))
   {
-    if (key->type == Key::FOREIGN_KEY &&
-        create_info->vers_info.check_unit == VERS_TIMESTAMP)
-    {
-      Foreign_key *fk_key= (Foreign_key*) key;
-      enum enum_fk_option op;
-      if (fk_modifies_child(op=fk_key->update_opt) ||
-          fk_modifies_child(op=fk_key->delete_opt))
-      {
-        my_error(ER_VERS_NOT_SUPPORTED, MYF(0), fk_option_name(op)->str,
-                 "TIMESTAMP(6) AS ROW START/END");
-        return true;
-      }
-    }
     if (key->type != Key::PRIMARY && key->type != Key::UNIQUE)
       continue;
 

storage/innobase/dict/dict0mem.cc

@@ -1489,16 +1489,12 @@ dict_index_t::vers_history_row(
 	ut_ad(col.vers_sys_end());
 	ulint nfield = dict_col_get_clust_pos(&col, this);
 	const byte *data = rec_get_nth_field(rec, offsets, nfield, &len);
-	if (col.mtype == DATA_FIXBINARY) {
+	if (col.vers_native()) {
-		ut_ad(len == sizeof timestamp_max_bytes);
-		return 0 != memcmp(data, timestamp_max_bytes, len);
-	} else {
-		ut_ad(col.mtype == DATA_INT);
 		ut_ad(len == sizeof trx_id_max_bytes);
 		return 0 != memcmp(data, trx_id_max_bytes, len);
 	}
-	ut_ad(0);
+	ut_ad(len == sizeof timestamp_max_bytes);
-	return false;
+	return 0 != memcmp(data, timestamp_max_bytes, len);
 }
 
 /** Check if record in secondary index is historical row.

storage/innobase/include/data0type.h

@@ -542,8 +542,6 @@ struct dtype_t{
 	unsigned	mbmaxlen:3;	/*!< maximum length of a character,
 					in bytes */
 
-	/** @return whether this is system field */
-	bool vers_sys_field() const { return prtype & DATA_VERSIONED; }
 	/** @return whether this is system versioned user field */
 	bool is_versioned() const { return !(~prtype & DATA_VERSIONED); }
 	/** @return whether this is the system field start */

storage/innobase/include/dict0mem.h

@@ -605,8 +605,13 @@ struct dict_col_t{
 	/** @return whether NULL is an allowed value for this column */
 	bool is_nullable() const { return !(prtype & DATA_NOT_NULL); }
 
-	/** @return whether this is system field */
+	/** @return whether table of this system field is TRX_ID-based */
-	bool vers_sys_field() const { return prtype & DATA_VERSIONED; }
+	bool vers_native() const
+	{
+		ut_ad(vers_sys_start() || vers_sys_end());
+		ut_ad(mtype == DATA_INT || mtype == DATA_FIXBINARY);
+		return mtype == DATA_INT;
+	}
 	/** @return whether this is system versioned */
 	bool is_versioned() const { return !(~prtype & DATA_VERSIONED); }
 	/** @return whether this is the system version start */

storage/innobase/include/row0upd.h

@@ -474,11 +474,16 @@ struct upd_t{
 		return(false);
 	}
 
-	/** Determine if the update affects a system versioned column. */
+	/** Determine if the update affects a system versioned column or row_end. */
 	bool affects_versioned() const
 	{
 		for (ulint i = 0; i < n_fields; i++) {
-			if (fields[i].new_val.type.vers_sys_field()) {
+			dtype_t type = fields[i].new_val.type;
+			if (type.is_versioned()) {
+				return true;
+			}
+			// versioned DELETE is UPDATE SET row_end=NOW
+			if (type.vers_sys_end()) {
 				return true;
 			}
 		}
@@ -563,6 +568,12 @@ struct upd_node_t{
 	dtuple_t*	row;	/*!< NULL, or a copy (also fields copied to
 				heap) of the row to update; this must be reset
 				to NULL after a successful update */
+	dtuple_t*	historical_row;	/*!< historical row used in
+				CASCADE UPDATE/SET NULL;
+				allocated from historical_heap  */
+	mem_heap_t*	historical_heap; /*!< heap for historical row insertion;
+				created when row to update is located;
+				freed right before row update */
 	row_ext_t*	ext;	/*!< NULL, or prefixes of the externally
 				stored columns in the old row */
 	dtuple_t*	upd_row;/* NULL, or a copy of the updated row */
@@ -577,6 +588,22 @@ struct upd_node_t{
 				/* column assignment list */
 	ulint		magic_n;
 
+	/** Also set row_start = CURRENT_TIMESTAMP/trx->id
+	@param[in]	trx	transaction */
+	void make_versioned_update(const trx_t* trx);
+	/** Only set row_end = CURRENT_TIMESTAMP/trx->id.
+	Do not touch other fields at all.
+	@param[in]	trx	transaction */
+	void make_versioned_delete(const trx_t* trx);
+
+private:
+	/** Appends row_start or row_end field to update vector and sets a
+	CURRENT_TIMESTAMP/trx->id value to it.
+	Supposed to be called only by make_versioned_update() and
+	make_versioned_delete().
+	@param[in]	trx	transaction
+	@param[in]	vers_sys_idx	table->row_start or table->row_end */
+	void make_versioned_helper(const trx_t* trx, ulint idx);
 };
 
 #define	UPD_NODE_MAGIC_N	1579975

storage/innobase/row/row0ins.cc

@@ -553,6 +553,8 @@ row_ins_cascade_calc_update_vec(
 				ufield->exp = NULL;
 
 				ufield->new_val = parent_ufield->new_val;
+				dfield_get_type(&ufield->new_val)->prtype |=
+					col->prtype & DATA_VERSIONED;
 				ufield_len = dfield_get_len(&ufield->new_val);
 
 				/* Clear the "external storage" flag */
@@ -1391,6 +1393,15 @@ row_ins_foreign_check_on_constraint(
 		}
 	}
 
+	if (table->versioned() && cascade->is_delete != PLAIN_DELETE
+	    && cascade->update->affects_versioned()) {
+		ut_ad(!cascade->historical_heap);
+		cascade->historical_heap = mem_heap_create(128);
+		cascade->historical_row = row_build(
+			ROW_COPY_POINTERS, clust_index, clust_rec, NULL, table,
+			NULL, NULL, NULL, cascade->historical_heap);
+	}
+
 	/* Store pcur position and initialize or store the cascade node
 	pcur stored position */
 
@@ -1613,6 +1624,19 @@ row_ins_check_foreign_constraint(
 		}
 	}
 
+	if (que_node_get_type(thr->run_node) == QUE_NODE_INSERT) {
+		ins_node_t* insert_node =
+			static_cast<ins_node_t*>(thr->run_node);
+		dict_table_t* table = insert_node->index->table;
+		if (table->versioned()) {
+			dfield_t* row_end = dtuple_get_nth_field(
+				insert_node->row, table->vers_end);
+			if (row_end->vers_history_row()) {
+				goto exit_func;
+			}
+		}
+	}
+
 	if (check_ref) {
 		check_table = foreign->referenced_table;
 		check_index = foreign->referenced_index;

storage/innobase/row/row0mysql.cc

@@ -1866,50 +1866,15 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 
 	ut_ad(!prebuilt->versioned_write || node->table->versioned());
 
-	bool vers_set_fields = prebuilt->versioned_write
+	if (prebuilt->versioned_write) {
-		&& (node->is_delete ? node->is_delete == VERSIONED_DELETE
+		if (node->is_delete == VERSIONED_DELETE) {
-		    : node->update->affects_versioned());
+			node->make_versioned_delete(trx);
+		} else if (node->update->affects_versioned()) {
+			node->make_versioned_update(trx);
+		}
+	}
 
 	for (;;) {
-		if (vers_set_fields) {
-			/* System Versioning: modify update vector to set
-			row_start (or row_end in case of DELETE)
-			to current trx_id. */
-			dict_table_t* table = node->table;
-			dict_index_t* clust_index = dict_table_get_first_index(table);
-			upd_t* uvect = node->update;
-			upd_field_t* ufield;
-			dict_col_t* col;
-			unsigned col_idx;
-			if (node->is_delete) {
-				ufield = &uvect->fields[0];
-				uvect->n_fields = 0;
-				node->is_delete = VERSIONED_DELETE;
-				col_idx = table->vers_end;
-			} else {
-				ut_ad(uvect->n_fields < table->n_cols);
-				ufield = &uvect->fields[uvect->n_fields];
-				col_idx = table->vers_start;
-			}
-			col = &table->cols[col_idx];
-			UNIV_MEM_INVALID(ufield, sizeof *ufield);
-			{
-				ulint field_no = dict_col_get_clust_pos(col, clust_index);
-				ut_ad(field_no != ULINT_UNDEFINED);
-				ufield->field_no = field_no;
-			}
-			ufield->orig_len = 0;
-			ufield->exp = NULL;
-
-			mach_write_to_8(node->update->vers_sys_value, trx->id);
-			dfield_t* dfield = &ufield->new_val;
-			dfield_set_data(dfield, node->update->vers_sys_value, 8);
-			dict_col_copy_type(col, &dfield->type);
-
-			uvect->n_fields++;
-			ut_ad(node->in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
-		}
-
 		thr->run_node = node;
 		thr->prev_node = node;
 		thr->fk_cascade_depth = 0;
@@ -2169,6 +2134,77 @@ row_mysql_unfreeze_data_dictionary(
 	trx->dict_operation_lock_mode = 0;
 }
 
+/** Write query start time as SQL field data to a buffer. Needed by InnoDB.
+@param	thd	Thread object
+@param	buf	Buffer to hold start time data */
+void thd_get_query_start_data(THD *thd, char *buf);
+
+/** Function restores btr_pcur_t, creates dtuple_t from rec_t,
+sets row_end = CURRENT_TIMESTAMP/trx->id, inserts it to a table and updates
+table statistics.
+This is used in UPDATE CASCADE/SET NULL of a system versioning table.
+@param[in]	thr	current query thread
+@param[in]	node	a node which just updated a row in a foreign table
+@return DB_SUCCESS or some error */
+static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
+{
+	const trx_t* trx = thr_get_trx(thr);
+	dict_table_t* table = node->table;
+	ut_ad(table->versioned());
+
+	dtuple_t* row = node->historical_row;
+	ut_ad(row);
+	node->historical_row = NULL;
+
+	ins_node_t* insert_node =
+		ins_node_create(INS_DIRECT, table, node->historical_heap);
+
+	ins_node_set_new_row(insert_node, row);
+
+	dfield_t* row_end = dtuple_get_nth_field(row, table->vers_end);
+	char row_end_data[8];
+	if (dict_table_get_nth_col(table, table->vers_end)->vers_native()) {
+		mach_write_to_8(row_end_data, trx->id);
+		dfield_set_data(row_end, row_end_data, 8);
+	} else {
+		thd_get_query_start_data(trx->mysql_thd, row_end_data);
+		dfield_set_data(row_end, row_end_data, 7);
+	}
+
+	for (;;) {
+		thr->run_node = insert_node;
+		thr->prev_node = insert_node;
+
+		row_ins_step(thr);
+
+		switch (trx->error_state) {
+		case DB_LOCK_WAIT:
+			que_thr_stop_for_mysql(thr);
+			lock_wait_suspend_thread(thr);
+
+			if (trx->error_state == DB_SUCCESS) {
+				continue;
+			}
+
+			/* fall through */
+		default:
+			/* Other errors are handled for the parent node. */
+			thr->fk_cascade_depth = 0;
+			goto exit;
+
+		case DB_SUCCESS:
+			srv_stats.n_rows_inserted.inc(
+				static_cast<size_t>(trx->id));
+			dict_stats_update_if_needed(table);
+			goto exit;
+		}
+	}
+exit:
+	mem_heap_free(node->historical_heap);
+	node->historical_heap = NULL;
+	return trx->error_state;
+}
+
 /**********************************************************************//**
 Does a cascaded delete or set null in a foreign key operation.
 @return error code or DB_SUCCESS */
@@ -2188,53 +2224,21 @@ row_update_cascade_for_mysql(
                 return(DB_FOREIGN_EXCEED_MAX_CASCADE);
         }
 
-	trx_t* trx = thr_get_trx(thr);
+	const trx_t* trx = thr_get_trx(thr);
 
-	bool vers_set_fields = node->table->versioned()
+	if (table->versioned()) {
-				  && (node->is_delete == PLAIN_DELETE
+		if (node->is_delete == PLAIN_DELETE) {
-				      || node->update->affects_versioned());
+			node->make_versioned_delete(trx);
+		} else if (node->update->affects_versioned()) {
+			dberr_t err = row_update_vers_insert(thr, node);
+			if (err != DB_SUCCESS) {
+				return err;
+			}
+			node->make_versioned_update(trx);
+		}
+	}
 
 	for (;;) {
-		if (vers_set_fields) {
-			// FIXME: code duplication with row_update_for_mysql()
-			/* System Versioning: modify update vector to set
-			row_start (or row_end in case of DELETE)
-			to current trx_id. */
-			dict_table_t* table = node->table;
-			dict_index_t* clust_index = dict_table_get_first_index(table);
-			upd_t* uvect = node->update;
-			upd_field_t* ufield;
-			dict_col_t* col;
-			unsigned col_idx;
-			if (node->is_delete) {
-				ufield = &uvect->fields[0];
-				uvect->n_fields = 0;
-				node->is_delete = VERSIONED_DELETE;
-				col_idx = table->vers_end;
-			} else {
-				ut_ad(uvect->n_fields < table->n_cols);
-				ufield = &uvect->fields[uvect->n_fields];
-				col_idx = table->vers_start;
-			}
-			col = &table->cols[col_idx];
-			UNIV_MEM_INVALID(ufield, sizeof *ufield);
-			{
-				ulint field_no = dict_col_get_clust_pos(col, clust_index);
-				ut_ad(field_no != ULINT_UNDEFINED);
-				ufield->field_no = field_no;
-			}
-			ufield->orig_len = 0;
-			ufield->exp = NULL;
-
-			mach_write_to_8(node->update->vers_sys_value, trx->id);
-			dfield_t* dfield = &ufield->new_val;
-			dfield_set_data(dfield, node->update->vers_sys_value, 8);
-			dict_col_copy_type(col, &dfield->type);
-
-			uvect->n_fields++;
-			ut_ad(node->in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
-		}
-
 		thr->run_node = node;
 		thr->prev_node = node;
 

storage/innobase/row/row0upd.cc

@@ -3448,3 +3448,57 @@ error_handling:
 
 	DBUG_RETURN(thr);
 }
+
+/** Write query start time as SQL field data to a buffer. Needed by InnoDB.
+@param	thd	Thread object
+@param	buf	Buffer to hold start time data */
+void thd_get_query_start_data(THD *thd, char *buf);
+
+/** Appends row_start or row_end field to update vector and sets a
+CURRENT_TIMESTAMP/trx->id value to it.
+Supposed to be called only by make_versioned_update() and
+make_versioned_delete().
+@param[in]	trx	transaction
+@param[in]	vers_sys_idx	table->row_start or table->row_end */
+void upd_node_t::make_versioned_helper(const trx_t* trx, ulint idx)
+{
+	ut_ad(in_mysql_interface); // otherwise needs to recalculate
+				   // node->cmpl_info
+	ut_ad(idx == table->vers_start || idx == table->vers_end);
+
+	dict_index_t* clust_index = dict_table_get_first_index(table);
+
+	update->n_fields++;
+	upd_field_t* ufield =
+		upd_get_nth_field(update, upd_get_n_fields(update) - 1);
+	const dict_col_t* col = dict_table_get_nth_col(table, idx);
+
+	upd_field_set_field_no(ufield, dict_col_get_clust_pos(col, clust_index),
+			       clust_index);
+
+	char* where = reinterpret_cast<char*>(update->vers_sys_value);
+	if (col->vers_native()) {
+		mach_write_to_8(where, trx->id);
+	} else {
+		thd_get_query_start_data(trx->mysql_thd, where);
+	}
+
+	dfield_set_data(&ufield->new_val, update->vers_sys_value, col->len);
+}
+
+/** Also set row_start = CURRENT_TIMESTAMP/trx->id
+@param[in]	trx	transaction */
+void upd_node_t::make_versioned_update(const trx_t* trx)
+{
+	make_versioned_helper(trx, table->vers_start);
+}
+
+/** Only set row_end = CURRENT_TIMESTAMP/trx->id.
+Do not touch other fields at all.
+@param[in]	trx	transaction */
+void upd_node_t::make_versioned_delete(const trx_t* trx)
+{
+	update->n_fields = 0;
+	is_delete = VERSIONED_DELETE;
+	make_versioned_helper(trx, table->vers_end);
+}

storage/innobase/trx/trx0rec.cc

@@ -2089,8 +2089,8 @@ trx_undo_report_row_operation(
 				if (!time.is_versioned()
 				    && index->table->versioned_by_id()
 				    && (!rec /* INSERT */
-					|| !update /* DELETE */
+					|| (update
-					|| update->affects_versioned())) {
+					    && update->affects_versioned()))) {
 					time.set_versioned(limit);
 				}
 			}