From 0cf97ad5b9f5ad0e4f4a3790755dd6e4d8df96dd Mon Sep 17 00:00:00 2001
From: Aleksey Midenkov <midenok@gmail.com>
Date: Sat, 13 Jan 2018 00:19:16 +0300
Subject: [PATCH 1/5] IB: CASCADE operation for DELETE

* Removed "Not supported for TIMESTAMP-based" error
* Fixed code duplication with node->vers_set_fields()
* Recovered foreign.test

[closes tempesta-tech#473]
---
 mysql-test/suite/versioning/common.inc        |  22 +++
 mysql-test/suite/versioning/common_finish.inc |   2 +
 mysql-test/suite/versioning/r/foreign.result  |  98 +++++++++----
 mysql-test/suite/versioning/t/foreign.test    |  50 +++----
 sql/sql_class.cc                              |  14 ++
 sql/sql_table.cc                              |  13 --
 storage/innobase/dict/dict0mem.cc             |  10 +-
 storage/innobase/include/dict0mem.h           |   7 +
 storage/innobase/include/row0upd.h            |   3 +
 storage/innobase/row/row0mysql.cc             | 130 ++++++++----------
 10 files changed, 198 insertions(+), 151 deletions(-)

diff --git a/mysql-test/suite/versioning/common.inc b/mysql-test/suite/versioning/common.inc
index 4324dd27cd8..aa3877d1570 100644
--- a/mysql-test/suite/versioning/common.inc
+++ b/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
diff --git a/mysql-test/suite/versioning/common_finish.inc b/mysql-test/suite/versioning/common_finish.inc
index 6894267f353..d753fc138fb 100644
--- a/mysql-test/suite/versioning/common_finish.inc
+++ b/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
diff --git a/mysql-test/suite/versioning/r/foreign.result b/mysql-test/suite/versioning/r/foreign.result
index bb9042fff9f..38c93d2611b 100644
--- a/mysql-test/suite/versioning/r/foreign.result
+++ b/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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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,37 @@ id int unique key
 ) engine innodb;
 create table child(
 parent_id int,
-sys_start timestamp(6) as row start invisible,
-sys_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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
+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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start invisible,
+sys_end SYS_DATATYPE as row end invisible,
 period for system_time(sys_start, sys_end)
 ) with system versioning
 engine innodb;
@@ -97,8 +115,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_end timestamp(6) as row end invisible,
+row_start SYS_DATATYPE as row start 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 +124,64 @@ 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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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
+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
+NULL	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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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 +212,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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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_end timestamp(6) as row end invisible,
+sys_start SYS_DATATYPE as row start 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`
diff --git a/mysql-test/suite/versioning/t/foreign.test b/mysql-test/suite/versioning/t/foreign.test
index 040d7564360..bd6cfd30522 100644
--- a/mysql-test/suite/versioning/t/foreign.test
+++ b/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
-eval select *, row_start < row_end, row_end < $sys_datatype_max from child for system_time all;
+select *, check_row(row_start, row_end) 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
-eval create table child(
+--replace_result $sys_datatype_expl SYS_DATATYPE
+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
-if (0)
-{
-update parent set id=id+1;
+update parent set id= id + 1;
 select * from child;
-select * from child for system_time from timestamp 0 to timestamp now(6);
-}
-## FIXME END
+select *, current_row(sys_end) as current_row from child for system_time all order by sys_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),
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index b3b2c8233bf..fb7399feec7 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -5198,6 +5198,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.
diff --git a/sql/sql_table.cc b/sql/sql_table.cc
index 0201a904058..d1af8d590da 100644
--- a/sql/sql_table.cc
+++ b/sql/sql_table.cc
@@ -4469,19 +4469,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;
 
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index 2e175731d5f..11936616cd2 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -1492,16 +1492,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) {
-		ut_ad(len == sizeof timestamp_max_bytes);
-		return 0 != memcmp(data, timestamp_max_bytes, len);
-	} else {
-		ut_ad(col.mtype == DATA_INT);
+	if (col.vers_native()) {
 		ut_ad(len == sizeof trx_id_max_bytes);
 		return 0 != memcmp(data, trx_id_max_bytes, len);
 	}
-	ut_ad(0);
-	return false;
+	ut_ad(len == sizeof timestamp_max_bytes);
+	return 0 != memcmp(data, timestamp_max_bytes, len);
 }
 
 /** Check if record in secondary index is historical row.
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 0bab513d051..d962c69c0c8 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -654,6 +654,13 @@ struct dict_col_t{
 
 	/** @return whether this is system field */
 	bool vers_sys_field() const { return prtype & DATA_VERSIONED; }
+	/** @return whether table of this system field is TRX_ID-based */
+	bool vers_native() const
+	{
+		ut_ad(vers_sys_field());
+		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 */
diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index 01fc6cda6ae..d010e15eab7 100644
--- a/storage/innobase/include/row0upd.h
+++ b/storage/innobase/include/row0upd.h
@@ -577,6 +577,9 @@ struct upd_node_t{
 				/* column assignment list */
 	ulint		magic_n;
 
+	/** System Versioning: modify update vector to set row_start
+	 * (or row_end in case of DELETE) to current trx_id. */
+	void		vers_set_fields(const trx_t* trx);
 };
 
 #define	UPD_NODE_MAGIC_N	1579975
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 835ca2ec2d1..a47f1a9b239 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -71,6 +71,15 @@ Created 9/17/2000 Heikki Tuuri
 #include <deque>
 #include <vector>
 
+/**
+  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);
+
+
 /** Provide optional 4.x backwards compatibility for 5.0 and above */
 ibool	row_rollback_on_timeout	= FALSE;
 
@@ -1781,6 +1790,50 @@ init_fts_doc_id_for_ref(
 	}
 }
 
+/** System Versioning: modify update vector to set row_start
+ * (or row_end in case of DELETE) to current trx_id. */
+void upd_node_t::vers_set_fields(const trx_t* trx)
+{
+	dict_index_t* clust_index = dict_table_get_first_index(table);
+	upd_field_t* ufield;
+	dict_col_t* col;
+	unsigned col_idx;
+	if (is_delete) {
+		ufield = &update->fields[0];
+		update->n_fields = 0;
+		is_delete = VERSIONED_DELETE;
+		col_idx = table->vers_end;
+	} else {
+		ut_ad(update->n_fields < table->n_cols);
+		ufield = &update->fields[update->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;
+
+	if (col->vers_native())
+	{
+		mach_write_to_8(update->vers_sys_value, trx->id);
+	} else {
+		thd_get_query_start_data(trx->mysql_thd, (char *)
+					update->vers_sys_value);
+	}
+
+	dfield_t* dfield = &ufield->new_val;
+	dfield_set_data(dfield, update->vers_sys_value, col->len);
+	dict_col_copy_type(col, &dfield->type);
+
+	update->n_fields++;
+	ut_ad(in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
+}
+
 /** Does an update or delete of a row for MySQL.
 @param[in,out]	prebuilt	prebuilt struct in MySQL handle
 @return error code or DB_SUCCESS */
@@ -1877,42 +1930,7 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 
 	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
+			node->vers_set_fields(trx);
 		}
 
 		thr->run_node = node;
@@ -2193,7 +2211,7 @@ 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()
 				  && (node->is_delete == PLAIN_DELETE
@@ -2201,43 +2219,7 @@ row_update_cascade_for_mysql(
 
 	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
+			node->vers_set_fields(trx);
 		}
 
 		thr->run_node = node;

From fd73c6dda4dcc8cfff494e296c1cef21dfd64660 Mon Sep 17 00:00:00 2001
From: Eugene Kosov <claprix@yandex.ru>
Date: Thu, 22 Feb 2018 19:14:58 +0300
Subject: [PATCH 2/5] Vers IB: Mark unversioned fields in system versioning
 tables

Some fields in system-versioned table may be unversioned.
SQL layer marks unversioned.
And this patch makes InnoDB mark unversioned too because of two reasons:
1) by default fields are versioned
2) most of fields are expected to be versioned

dtype_t::vers_sys_field(): fixed return true on row_start/row_end
dict_col_t::vers_sys_field(): fixed return true on row_start/row_end
---
 storage/innobase/dict/dict0mem.cc         | 17 +++++++++--------
 storage/innobase/handler/ha_innodb.cc     |  6 +++---
 storage/innobase/handler/handler0alter.cc |  6 +++---
 storage/innobase/include/data0type.h      | 14 ++++++++------
 storage/innobase/include/dict0mem.h       | 11 +++++++----
 storage/innobase/include/row0upd.h        |  9 +++++++--
 storage/innobase/trx/trx0rec.cc           |  4 ++--
 7 files changed, 39 insertions(+), 28 deletions(-)

diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index 11936616cd2..f96b87995f8 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -312,14 +312,15 @@ dict_mem_table_add_col(
 
 	dict_mem_fill_column_struct(col, i, mtype, prtype, len);
 
-	switch (prtype & DATA_VERSIONED) {
-	case DATA_VERS_START:
-		ut_ad(!table->vers_start);
-		table->vers_start = i;
-		break;
-	case DATA_VERS_END:
-		ut_ad(!table->vers_end);
-		table->vers_end = i;
+	if ((prtype & DATA_UNVERSIONED) != DATA_UNVERSIONED) {
+		if (prtype & DATA_VERS_START) {
+			ut_ad(!table->vers_start);
+			table->vers_start = i;
+		}
+		if (prtype & DATA_VERS_END) {
+			ut_ad(!table->vers_end);
+			table->vers_end = i;
+		}
 	}
 }
 
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index b4f3a27b5af..3c12d05757b 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -11265,9 +11265,9 @@ create_table_info_t::create_table_def()
 				vers_row = DATA_VERS_START;
 			} else if (i == m_form->s->row_end_field) {
 				vers_row = DATA_VERS_END;
-			} else if (!(field->flags
-				     & VERS_UPDATE_UNVERSIONED_FLAG)) {
-				vers_row = DATA_VERSIONED;
+			} else if (field->flags
+				   & VERS_UPDATE_UNVERSIONED_FLAG) {
+				vers_row = DATA_UNVERSIONED;
 			}
 		}
 
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index b775bfa40ee..79b96770553 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -5009,9 +5009,9 @@ new_clustered_failed:
 				} else if (i ==
 					   altered_table->s->row_end_field) {
 					field_type |= DATA_VERS_END;
-				} else if (!(field->flags
-					     & VERS_UPDATE_UNVERSIONED_FLAG)) {
-					field_type |= DATA_VERSIONED;
+				} else if (field->flags
+					   & VERS_UPDATE_UNVERSIONED_FLAG) {
+					field_type |= DATA_UNVERSIONED;
 				}
 			}
 
diff --git a/storage/innobase/include/data0type.h b/storage/innobase/include/data0type.h
index 7e1c362cf8d..897ad19f71e 100644
--- a/storage/innobase/include/data0type.h
+++ b/storage/innobase/include/data0type.h
@@ -192,8 +192,7 @@ be less than 256 */
 /** System Versioning */
 #define DATA_VERS_START	16384U	/* start system field */
 #define DATA_VERS_END	32768U	/* end system field */
-/** system-versioned user data column */
-#define DATA_VERSIONED (DATA_VERS_START|DATA_VERS_END)
+#define DATA_UNVERSIONED (DATA_VERS_START|DATA_VERS_END) /* unversioned user field */
 
 /** Check whether locking is disabled (never). */
 #define dict_table_is_locking_disabled(table) false
@@ -543,18 +542,21 @@ struct dtype_t{
 					in bytes */
 
 	/** @return whether this is system field */
-	bool vers_sys_field() const { return prtype & DATA_VERSIONED; }
+	bool vers_sys_field() const
+	{
+		return vers_sys_start() || vers_sys_end();
+	}
 	/** @return whether this is system versioned user field */
-	bool is_versioned() const { return !(~prtype & DATA_VERSIONED); }
+	bool is_versioned() const { return (prtype & DATA_UNVERSIONED) == 0; }
 	/** @return whether this is the system field start */
 	bool vers_sys_start() const
 	{
-		return (prtype & DATA_VERSIONED) == DATA_VERS_START;
+		return (prtype & DATA_UNVERSIONED) == DATA_VERS_START;
 	}
 	/** @return whether this is the system field end */
 	bool vers_sys_end() const
 	{
-		return (prtype & DATA_VERSIONED) == DATA_VERS_END;
+		return (prtype & DATA_UNVERSIONED) == DATA_VERS_END;
 	}
 };
 
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index d962c69c0c8..6b470247cf2 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -653,7 +653,10 @@ struct dict_col_t{
 	bool is_nullable() const { return !(prtype & DATA_NOT_NULL); }
 
 	/** @return whether this is system field */
-	bool vers_sys_field() const { return prtype & DATA_VERSIONED; }
+	bool vers_sys_field() const
+	{
+		return vers_sys_start() || vers_sys_end();
+	}
 	/** @return whether table of this system field is TRX_ID-based */
 	bool vers_native() const
 	{
@@ -662,16 +665,16 @@ struct dict_col_t{
 		return mtype == DATA_INT;
 	}
 	/** @return whether this is system versioned */
-	bool is_versioned() const { return !(~prtype & DATA_VERSIONED); }
+	bool is_versioned() const { return (prtype & DATA_UNVERSIONED) == 0; }
 	/** @return whether this is the system version start */
 	bool vers_sys_start() const
 	{
-		return (prtype & DATA_VERSIONED) == DATA_VERS_START;
+		return (prtype & DATA_UNVERSIONED) == DATA_VERS_START;
 	}
 	/** @return whether this is the system version end */
 	bool vers_sys_end() const
 	{
-		return (prtype & DATA_VERSIONED) == DATA_VERS_END;
+		return (prtype & DATA_UNVERSIONED) == DATA_VERS_END;
 	}
 
 	/** @return whether this is an instantly-added column */
diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index d010e15eab7..213cd42a297 100644
--- a/storage/innobase/include/row0upd.h
+++ b/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;
 			}
 		}
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index f3f04ab5e1c..41d9b1a5fad 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/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->affects_versioned())) {
+					|| (update
+					    && update->affects_versioned()))) {
 					time.set_versioned(limit);
 				}
 			}

From 19a182b1ea5b688aeb157cee2675a883b4e9fd34 Mon Sep 17 00:00:00 2001
From: Eugene Kosov <claprix@yandex.ru>
Date: Mon, 19 Mar 2018 17:25:58 +0300
Subject: [PATCH 3/5] MDEV-15364 FOREIGN CASCADE operations in system versioned
 referenced tables

Make foreign system versioning tables work in CASCADE UPDATE/SET NULL.
In that case basically row update is performed. This patch makes insert
of a historical row performed too.

row_update_versioned_insert(): restores btr_pcur_t, reads row from it, makes
row historical and inserts to table.

row_ins_check_foreign_constraint(): disable constraint check for historical
rows because it has no sense. Also check will fail always, because referenced
table is updated at that point.

row_update_cascade_for_mysql(): insert historical row for system versioning
tables before updating current row.

revert DATA_VERSIONED -> DATA_UNVERSIONED
---
 mysql-test/suite/versioning/r/foreign.result |  59 ++++++++
 mysql-test/suite/versioning/t/foreign.test   |  51 +++++++
 storage/innobase/dict/dict0mem.cc            |  17 +--
 storage/innobase/handler/ha_innodb.cc        |   6 +-
 storage/innobase/handler/handler0alter.cc    |   6 +-
 storage/innobase/include/data0type.h         |  14 +-
 storage/innobase/include/dict0mem.h          |  13 +-
 storage/innobase/include/row0upd.h           |  21 ++-
 storage/innobase/row/row0ins.cc              |  22 +++
 storage/innobase/row/row0mysql.cc            | 150 ++++++++++---------
 storage/innobase/row/row0upd.cc              |  54 +++++++
 11 files changed, 310 insertions(+), 103 deletions(-)

diff --git a/mysql-test/suite/versioning/r/foreign.result b/mysql-test/suite/versioning/r/foreign.result
index 38c93d2611b..edf5632f027 100644
--- a/mysql-test/suite/versioning/r/foreign.result
+++ b/mysql-test/suite/versioning/r/foreign.result
@@ -82,6 +82,7 @@ parent_id
 select * from child for system_time all;
 parent_id
 1
+1
 2
 drop table child;
 drop table parent;
@@ -160,6 +161,7 @@ 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);
@@ -171,7 +173,9 @@ 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;
@@ -231,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;
diff --git a/mysql-test/suite/versioning/t/foreign.test b/mysql-test/suite/versioning/t/foreign.test
index bd6cfd30522..566d481c2a8 100644
--- a/mysql-test/suite/versioning/t/foreign.test
+++ b/mysql-test/suite/versioning/t/foreign.test
@@ -266,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
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index f96b87995f8..11936616cd2 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -312,15 +312,14 @@ dict_mem_table_add_col(
 
 	dict_mem_fill_column_struct(col, i, mtype, prtype, len);
 
-	if ((prtype & DATA_UNVERSIONED) != DATA_UNVERSIONED) {
-		if (prtype & DATA_VERS_START) {
-			ut_ad(!table->vers_start);
-			table->vers_start = i;
-		}
-		if (prtype & DATA_VERS_END) {
-			ut_ad(!table->vers_end);
-			table->vers_end = i;
-		}
+	switch (prtype & DATA_VERSIONED) {
+	case DATA_VERS_START:
+		ut_ad(!table->vers_start);
+		table->vers_start = i;
+		break;
+	case DATA_VERS_END:
+		ut_ad(!table->vers_end);
+		table->vers_end = i;
 	}
 }
 
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 3c12d05757b..b4f3a27b5af 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -11265,9 +11265,9 @@ create_table_info_t::create_table_def()
 				vers_row = DATA_VERS_START;
 			} else if (i == m_form->s->row_end_field) {
 				vers_row = DATA_VERS_END;
-			} else if (field->flags
-				   & VERS_UPDATE_UNVERSIONED_FLAG) {
-				vers_row = DATA_UNVERSIONED;
+			} else if (!(field->flags
+				     & VERS_UPDATE_UNVERSIONED_FLAG)) {
+				vers_row = DATA_VERSIONED;
 			}
 		}
 
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 79b96770553..b775bfa40ee 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -5009,9 +5009,9 @@ new_clustered_failed:
 				} else if (i ==
 					   altered_table->s->row_end_field) {
 					field_type |= DATA_VERS_END;
-				} else if (field->flags
-					   & VERS_UPDATE_UNVERSIONED_FLAG) {
-					field_type |= DATA_UNVERSIONED;
+				} else if (!(field->flags
+					     & VERS_UPDATE_UNVERSIONED_FLAG)) {
+					field_type |= DATA_VERSIONED;
 				}
 			}
 
diff --git a/storage/innobase/include/data0type.h b/storage/innobase/include/data0type.h
index 897ad19f71e..2eaa9042e9c 100644
--- a/storage/innobase/include/data0type.h
+++ b/storage/innobase/include/data0type.h
@@ -192,7 +192,8 @@ be less than 256 */
 /** System Versioning */
 #define DATA_VERS_START	16384U	/* start system field */
 #define DATA_VERS_END	32768U	/* end system field */
-#define DATA_UNVERSIONED (DATA_VERS_START|DATA_VERS_END) /* unversioned user field */
+/** system-versioned user data column */
+#define DATA_VERSIONED (DATA_VERS_START|DATA_VERS_END)
 
 /** Check whether locking is disabled (never). */
 #define dict_table_is_locking_disabled(table) false
@@ -541,22 +542,17 @@ 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 vers_sys_start() || vers_sys_end();
-	}
 	/** @return whether this is system versioned user field */
-	bool is_versioned() const { return (prtype & DATA_UNVERSIONED) == 0; }
+	bool is_versioned() const { return !(~prtype & DATA_VERSIONED); }
 	/** @return whether this is the system field start */
 	bool vers_sys_start() const
 	{
-		return (prtype & DATA_UNVERSIONED) == DATA_VERS_START;
+		return (prtype & DATA_VERSIONED) == DATA_VERS_START;
 	}
 	/** @return whether this is the system field end */
 	bool vers_sys_end() const
 	{
-		return (prtype & DATA_UNVERSIONED) == DATA_VERS_END;
+		return (prtype & DATA_VERSIONED) == DATA_VERS_END;
 	}
 };
 
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 6b470247cf2..1f978224617 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -652,29 +652,24 @@ 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 */
-	bool vers_sys_field() const
-	{
-		return vers_sys_start() || vers_sys_end();
-	}
 	/** @return whether table of this system field is TRX_ID-based */
 	bool vers_native() const
 	{
-		ut_ad(vers_sys_field());
+		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_UNVERSIONED) == 0; }
+	bool is_versioned() const { return !(~prtype & DATA_VERSIONED); }
 	/** @return whether this is the system version start */
 	bool vers_sys_start() const
 	{
-		return (prtype & DATA_UNVERSIONED) == DATA_VERS_START;
+		return (prtype & DATA_VERSIONED) == DATA_VERS_START;
 	}
 	/** @return whether this is the system version end */
 	bool vers_sys_end() const
 	{
-		return (prtype & DATA_UNVERSIONED) == DATA_VERS_END;
+		return (prtype & DATA_VERSIONED) == DATA_VERS_END;
 	}
 
 	/** @return whether this is an instantly-added column */
diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index 213cd42a297..2a185bd9d88 100644
--- a/storage/innobase/include/row0upd.h
+++ b/storage/innobase/include/row0upd.h
@@ -568,6 +568,8 @@ 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 */
 	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 */
@@ -582,9 +584,22 @@ struct upd_node_t{
 				/* column assignment list */
 	ulint		magic_n;
 
-	/** System Versioning: modify update vector to set row_start
-	 * (or row_end in case of DELETE) to current trx_id. */
-	void		vers_set_fields(const trx_t* trx);
+	/** 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
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index e87e857c812..13c64ced5f7 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/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,13 @@ row_ins_foreign_check_on_constraint(
 		}
 	}
 
+	if (table->versioned() && cascade->is_delete != PLAIN_DELETE
+	    && cascade->update->affects_versioned()) {
+		cascade->historical_row =
+			row_build(ROW_COPY_DATA, clust_index, clust_rec, NULL,
+				  table, NULL, NULL, NULL, thr->prebuilt->heap);
+	}
+
 	/* Store pcur position and initialize or store the cascade node
 	pcur stored position */
 
@@ -1613,6 +1622,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;
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index a47f1a9b239..631fca7959f 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -71,15 +71,6 @@ Created 9/17/2000 Heikki Tuuri
 #include <deque>
 #include <vector>
 
-/**
-  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);
-
-
 /** Provide optional 4.x backwards compatibility for 5.0 and above */
 ibool	row_rollback_on_timeout	= FALSE;
 
@@ -1790,50 +1781,6 @@ init_fts_doc_id_for_ref(
 	}
 }
 
-/** System Versioning: modify update vector to set row_start
- * (or row_end in case of DELETE) to current trx_id. */
-void upd_node_t::vers_set_fields(const trx_t* trx)
-{
-	dict_index_t* clust_index = dict_table_get_first_index(table);
-	upd_field_t* ufield;
-	dict_col_t* col;
-	unsigned col_idx;
-	if (is_delete) {
-		ufield = &update->fields[0];
-		update->n_fields = 0;
-		is_delete = VERSIONED_DELETE;
-		col_idx = table->vers_end;
-	} else {
-		ut_ad(update->n_fields < table->n_cols);
-		ufield = &update->fields[update->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;
-
-	if (col->vers_native())
-	{
-		mach_write_to_8(update->vers_sys_value, trx->id);
-	} else {
-		thd_get_query_start_data(trx->mysql_thd, (char *)
-					update->vers_sys_value);
-	}
-
-	dfield_t* dfield = &ufield->new_val;
-	dfield_set_data(dfield, update->vers_sys_value, col->len);
-	dict_col_copy_type(col, &dfield->type);
-
-	update->n_fields++;
-	ut_ad(in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
-}
-
 /** Does an update or delete of a row for MySQL.
 @param[in,out]	prebuilt	prebuilt struct in MySQL handle
 @return error code or DB_SUCCESS */
@@ -1924,15 +1871,15 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 
 	ut_ad(!prebuilt->versioned_write || node->table->versioned());
 
-	bool vers_set_fields = prebuilt->versioned_write
-		&& (node->is_delete ? node->is_delete == VERSIONED_DELETE
-		    : node->update->affects_versioned());
+	if (prebuilt->versioned_write) {
+		if (node->is_delete == VERSIONED_DELETE) {
+			node->make_versioned_delete(trx);
+		} else if (node->update->affects_versioned()) {
+			node->make_versioned_update(trx);
+		}
+	}
 
 	for (;;) {
-		if (vers_set_fields) {
-			node->vers_set_fields(trx);
-		}
-
 		thr->run_node = node;
 		thr->prev_node = node;
 		thr->fk_cascade_depth = 0;
@@ -2192,6 +2139,71 @@ 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, thr->prebuilt->heap);
+
+	ins_node_set_new_row(insert_node, row);
+
+	dfield_t* row_end = dtuple_get_nth_field(row, table->vers_end);
+	char* where = static_cast<char*>(dfield_get_data(row_end));
+	if (dict_table_get_nth_col(table, table->vers_end)->vers_native()) {
+		mach_write_to_8(where, trx->id);
+	} else {
+		thd_get_query_start_data(trx->mysql_thd, where);
+	}
+
+	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;
+			return trx->error_state;
+
+		case DB_SUCCESS:
+			srv_stats.n_rows_inserted.inc(
+				static_cast<size_t>(trx->id));
+			dict_stats_update_if_needed(table);
+			return DB_SUCCESS;
+		}
+	}
+}
+
 /**********************************************************************//**
 Does a cascaded delete or set null in a foreign key operation.
 @return error code or DB_SUCCESS */
@@ -2213,15 +2225,19 @@ row_update_cascade_for_mysql(
 
 	const trx_t* trx = thr_get_trx(thr);
 
-	bool vers_set_fields = node->table->versioned()
-				  && (node->is_delete == PLAIN_DELETE
-				      || node->update->affects_versioned());
+	if (table->versioned()) {
+		if (node->is_delete == PLAIN_DELETE) {
+			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) {
-			node->vers_set_fields(trx);
-		}
-
 		thr->run_node = node;
 		thr->prev_node = node;
 
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index e9002d6b41d..8c280c19615 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/storage/innobase/row/row0upd.cc
@@ -3454,3 +3454,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);
+}

From e41ae769c13699505cebbb99ab5cf9bf81fe39cc Mon Sep 17 00:00:00 2001
From: Eugene Kosov <claprix@yandex.ru>
Date: Fri, 6 Apr 2018 17:20:45 +0300
Subject: [PATCH 4/5] use dedicated heap for upd_node_t::historical_row and
 corresponding stuff to make its lifetime as short as possible

for historical row ROW_COPY_DATA -> ROW_COPY_POINTER for lower memory
consumption and better performance
---
 storage/innobase/include/row0upd.h |  6 +++++-
 storage/innobase/row/row0ins.cc    |  8 +++++---
 storage/innobase/row/row0mysql.cc  | 14 ++++++++++----
 3 files changed, 20 insertions(+), 8 deletions(-)

diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index 2a185bd9d88..e10f0906a8c 100644
--- a/storage/innobase/include/row0upd.h
+++ b/storage/innobase/include/row0upd.h
@@ -569,7 +569,11 @@ struct upd_node_t{
 				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 */
+				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 */
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 13c64ced5f7..876d8caec8a 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -1395,9 +1395,11 @@ row_ins_foreign_check_on_constraint(
 
 	if (table->versioned() && cascade->is_delete != PLAIN_DELETE
 	    && cascade->update->affects_versioned()) {
-		cascade->historical_row =
-			row_build(ROW_COPY_DATA, clust_index, clust_rec, NULL,
-				  table, NULL, NULL, NULL, thr->prebuilt->heap);
+		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
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 631fca7959f..6e343bb85fb 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2162,16 +2162,18 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 	node->historical_row = NULL;
 
 	ins_node_t* insert_node =
-		ins_node_create(INS_DIRECT, table, thr->prebuilt->heap);
+		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* where = static_cast<char*>(dfield_get_data(row_end));
+	char where[8];
 	if (dict_table_get_nth_col(table, table->vers_end)->vers_native()) {
 		mach_write_to_8(where, trx->id);
+		dfield_set_data(row_end, where, 8);
 	} else {
 		thd_get_query_start_data(trx->mysql_thd, where);
+		dfield_set_data(row_end, where, 7);
 	}
 
 	for (;;) {
@@ -2193,15 +2195,19 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 		default:
 			/* Other errors are handled for the parent node. */
 			thr->fk_cascade_depth = 0;
-			return trx->error_state;
+			goto exit;
 
 		case DB_SUCCESS:
 			srv_stats.n_rows_inserted.inc(
 				static_cast<size_t>(trx->id));
 			dict_stats_update_if_needed(table);
-			return DB_SUCCESS;
+			goto exit;
 		}
 	}
+exit:
+	mem_heap_free(node->historical_heap);
+	node->historical_heap = NULL;
+	return trx->error_state;
 }
 
 /**********************************************************************//**

From 7477b97e914651b6740b11ee6695f1bfb8772d43 Mon Sep 17 00:00:00 2001
From: Eugene Kosov <claprix@yandex.ru>
Date: Fri, 6 Apr 2018 17:38:14 +0300
Subject: [PATCH 5/5] rename "where" to "row_end_data"

---
 storage/innobase/row/row0mysql.cc | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 6e343bb85fb..643fd68f2fb 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2167,13 +2167,13 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 	ins_node_set_new_row(insert_node, row);
 
 	dfield_t* row_end = dtuple_get_nth_field(row, table->vers_end);
-	char where[8];
+	char row_end_data[8];
 	if (dict_table_get_nth_col(table, table->vers_end)->vers_native()) {
-		mach_write_to_8(where, trx->id);
-		dfield_set_data(row_end, where, 8);
+		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, where);
-		dfield_set_data(row_end, where, 7);
+		thd_get_query_start_data(trx->mysql_thd, row_end_data);
+		dfield_set_data(row_end, row_end_data, 7);
 	}
 
 	for (;;) {