IB: (0.4) foreign keys for versioned tables (#58)

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

+-- source include/have_innodb.inc
+
+#################
+# Test RESTRICT #
+#################
+
+create table parent(
+  id int unique key
+) engine innodb;
+
+create table child(
+  parent_id int,
+  foreign key(parent_id) references parent(id)
+    on delete restrict
+    on update restrict
+) engine innodb with system versioning;
+
+insert into parent values(1);
+insert into child values(1);
+
+-- error ER_ROW_IS_REFERENCED_2
+delete from parent where id = 1;
+delete from child where parent_id = 1;
+delete from parent where id = 1;
+
+insert into parent values(1);
+insert into child values(1);
+-- error ER_ROW_IS_REFERENCED_2
+update parent set id=id+1;
+delete from child;
+update parent set id=id+1;
+select * from child for system_time from timestamp '1-1-1' to timestamp now(6);
+
+drop table child;
+drop table parent;
+
+##############################################
+# Test when clustered index is a foreign key #
+##############################################
+
+create table parent(
+  id int(10) unsigned unique key
+) engine innodb;
+
+create table child(
+  parent_id int(10) unsigned primary key,
+  foreign key(parent_id) references parent(id)
+) engine innodb with system versioning;
+
+insert into parent values(1);
+insert into child values(1);
+
+-- error ER_ROW_IS_REFERENCED_2
+delete from parent where id = 1;
+
+drop table child;
+drop table parent;
+
+################
+# Test CASCADE #
+################
+
+create table parent(
+  id int unique key
+) engine innodb;
+
+create table child(
+  parent_id int,
+  foreign key(parent_id) references parent(id)
+    on delete cascade
+    on update cascade
+) engine innodb with system versioning;
+
+insert into parent values(1);
+insert into child values(1);
+
+delete from parent where id = 1;
+select * from child;
+select * from child for system_time from timestamp '1-1-1' to timestamp now(6);
+
+insert into parent values(1);
+insert into child values(1);
+update parent set id=id+1;
+select * from child;
+select * from child for system_time from timestamp '1-1-1' to timestamp now(6);
+
+drop table child;
+drop table parent;
+
+#################
+# Test SET NULL #
+#################
+
+create table parent(
+  id int unique key
+) engine innodb;
+
+create table child(
+  parent_id int,
+  foreign key(parent_id) references parent(id)
+    on delete set null
+    on update set null
+) engine innodb with system versioning;
+
+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;
+select * from child for system_time from timestamp '1-1-1' to timestamp now(6);
+delete from child;
+
+insert into parent values(1);
+insert into child values(1);
+update parent set id=id+1;
+select * from child;
+select * from child for system_time from timestamp '1-1-1' to timestamp now(6);
+
+drop table child;
+drop table parent;
+
+###########################
+# Parent table is foreign #
+###########################
+
+create or replace table parent(
+  id int unique key
+) engine innodb with system versioning;
+
+create or replace table child(
+  parent_id int,
+  foreign key(parent_id) references parent(id)
+) engine innodb;
+
+insert into parent values(1);
+insert into child values(1);
+-- error ER_ROW_IS_REFERENCED_2
+delete from parent;
+-- error ER_ROW_IS_REFERENCED_2
+update parent set id=2;
+
+delete from child;
+delete from parent;
+
+-- error ER_NO_REFERENCED_ROW_2
+insert into child values(1);
+
+insert into parent values(1);
+insert into child values(1);
+-- error ER_ROW_IS_REFERENCED_2
+delete from parent;
+-- error ER_ROW_IS_REFERENCED_2
+update parent set id=2;
+
+drop table child;
+drop table parent;

sql/share/errmsg-utf8.txt

@@ -7520,9 +7520,6 @@ ER_SYS_START_AND_SYS_END_SAME
 ER_GENERATED_FIELD_CANNOT_BE_SET_BY_USER
         eng "Generated field for System Versioning cannot be set by user"
 
-ER_FOREIGN_KEY_ON_SYSTEM_VERSIONED
-        eng "Foreign key clause is not yet supported in conjunction with system versioning"
-
 ER_UPDATE_INFO_WITH_SYSTEM_VERSIONING
         eng "Rows matched: %ld  Changed: %ld  Inserted: %ld  Warnings: %ld"
 

sql/sql_table.cc

@@ -4621,20 +4621,6 @@ handler *mysql_create_frm_image(THD *thd,
 
   if (create_info->versioned())
   {
-    // FIXME: This test doesn't detect foreign key relationship on the side of
-    //   parent table and System Time support will not work correctly for such
-    //   table either. But this cannot be implemented without changes to innodb
-    //   that are postponed for later time.
-    List_iterator_fast<Key> key_iterator(alter_info->key_list);
-    Key *key;
-    while ((key= key_iterator++))
-    {
-      if (key->type == Key::FOREIGN_KEY)
-      {
-        my_error(ER_FOREIGN_KEY_ON_SYSTEM_VERSIONED, MYF(0));
-        goto err;
-      }
-    }
     if(vers_prepare_keys(thd, create_info, alter_info, key_info,
                                 *key_count))
       goto err;

storage/innobase/handler/ha_innodb.cc

@@ -8960,6 +8960,7 @@ calc_row_difference(
 	buf = (byte*) upd_buff;
 
 	prebuilt->upd_node->versioned = false;
+	prebuilt->upd_node->vers_delete = false;
 
 	for (i = 0; i < table->s->fields; i++) {
 		field = table->field[i];

storage/innobase/include/row0upd.h

@@ -583,6 +583,7 @@ struct upd_node_t{
 				UPD_NODE_NO_ORD_CHANGE and
 				UPD_NODE_NO_SIZE_CHANGE, ORed */
 	bool		versioned;/* update is versioned */
+	bool		vers_delete;/* versioned delete */
 	/*----------------------*/
 	/* Local storage for this graph node */
 	ulint		state;	/*!< node execution state */

storage/innobase/row/row0ins.cc

@@ -1569,6 +1569,79 @@ class ib_dec_in_dtor {
 	ulint&		counter;
 };
 
+/*********************************************************************//**
+Reads sys_trx_end field from clustered index row.
+@return trx_id_t */
+static
+trx_id_t
+row_ins_get_sys_trx_end(
+/*===================================*/
+			const rec_t *rec,	/*!< in: clustered row */
+			ulint *offsets,		/*!< in: offsets */
+			dict_index_t *index)	/*!< in: clustered index */
+{
+	ut_a(dict_index_is_clust(index));
+
+	ulint len;
+	ulint nfield = dict_col_get_clust_pos(
+		&index->table->cols[index->table->vers_row_end], index);
+	const byte *field = rec_get_nth_field(rec, offsets, nfield, &len);
+	ut_a(len == 8);
+	return(mach_read_from_8(field));
+}
+
+/*********************************************************************//**
+Performs search at clustered index and returns sys_trx_end if row was found.
+@return DB_SUCCESS, DB_NO_REFERENCED_ROW */
+static
+dberr_t
+row_ins_search_sys_trx_end(
+/*=======================*/
+			dict_index_t *index,	/*!< in: index of record */
+			const rec_t *rec,	/*!< in: record */
+			trx_id_t *end_trx_id)	/*!< out: end_trx_id */
+{
+	rec_t *clust_rec;
+	bool found = false;
+	mem_heap_t *clust_heap = mem_heap_create(256);
+	ulint clust_offsets_[REC_OFFS_NORMAL_SIZE];
+	ulint *clust_offsets = clust_offsets_;
+	rec_offs_init(clust_offsets_);
+	btr_pcur_t clust_pcur;
+
+	dict_index_t *clust_index = dict_table_get_first_index(index->table);
+
+	dtuple_t *ref =
+		row_build_row_ref(ROW_COPY_POINTERS, index, rec, clust_heap);
+
+	mtr_t clust_mtr;
+	mtr_start(&clust_mtr);
+	btr_pcur_open_on_user_rec(clust_index, ref, PAGE_CUR_GE,
+				BTR_SEARCH_LEAF, &clust_pcur, &clust_mtr);
+
+	if (!btr_pcur_is_on_user_rec(&clust_pcur))
+		goto not_found;
+
+	clust_rec = btr_pcur_get_rec(&clust_pcur);
+	clust_offsets = rec_get_offsets(clust_rec, clust_index, clust_offsets,
+					ULINT_UNDEFINED, &clust_heap);
+	if (0 != cmp_dtuple_rec(ref, clust_rec, clust_offsets))
+		goto not_found;
+
+	*end_trx_id = row_ins_get_sys_trx_end(
+		clust_rec, clust_offsets, clust_index);
+	found = true;
+not_found:
+	mtr_commit(&clust_mtr);
+	btr_pcur_close(&clust_pcur);
+	mem_heap_free(clust_heap);
+	if (!found) {
+		fprintf(stderr, "InnoDB: foreign constraints: secondary index is out of sync\n");
+		return(DB_NO_REFERENCED_ROW);
+	}
+	return(DB_SUCCESS);
+}
+
 /***************************************************************//**
 Checks if foreign key constraint fails for an index entry. Sets shared locks
 which lock either the success or the failure of the constraint. NOTE that
@@ -1745,8 +1818,24 @@ row_ins_check_foreign_constraint(
 		cmp = cmp_dtuple_rec(entry, rec, offsets);
 
 		if (cmp == 0) {
-			if (rec_get_deleted_flag(rec,
-						 rec_offs_comp(offsets))) {
+			if (DICT_TF2_FLAG_IS_SET(check_table, DICT_TF2_VERSIONED)) {
+				trx_id_t end_trx_id = 0;
+
+				if (dict_index_is_clust(check_index)) {
+					end_trx_id =
+						row_ins_get_sys_trx_end(
+						rec, offsets, check_index);
+				} else if (row_ins_search_sys_trx_end(
+						check_index, rec, &end_trx_id) !=
+						DB_SUCCESS) {
+					break;
+				}
+
+				if (end_trx_id != TRX_ID_MAX)
+					continue;
+			}
+
+			if (rec_get_deleted_flag(rec, rec_offs_comp(offsets))) {
 				err = row_ins_set_shared_rec_lock(
 					LOCK_ORDINARY, block,
 					rec, check_index, offsets, thr);

storage/innobase/row/row0mysql.cc

@@ -1957,43 +1957,6 @@ row_update_for_mysql_using_upd_graph(
 					      prebuilt->clust_pcur);
 	}
 
-	if (DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED) &&
-		(node->is_delete || node->versioned))
-	{
-		/* System Versioning: modify update vector to set
-		   sys_trx_start (or sys_trx_end in case of DELETE)
-		   to current trx_id. */
-		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 = false;
-			col_idx = table->vers_row_end;
-		} else {
-			ut_ad(uvect->n_fields < node->table->n_cols);
-			ufield = &uvect->fields[uvect->n_fields];
-			col_idx = table->vers_row_start;
-		}
-		col = &table->cols[col_idx];
-		UNIV_MEM_INVALID(ufield, sizeof *ufield);
-		ufield->field_no = dict_col_get_clust_pos(col, clust_index);
-		ufield->orig_len = 0;
-		ufield->exp = NULL;
-
-		static const ulint fsize = sizeof(trx_id_t);
-		byte* buf = static_cast<byte*>(mem_heap_alloc(node->heap, fsize));
-		mach_write_to_8(buf, trx->id);
-		dfield_t* dfield = &ufield->new_val;
-		dfield_set_data(dfield, buf, fsize);
-		dict_col_copy_type(col, &dfield->type);
-
-		uvect->n_fields++;
-		ut_ad(node->in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
-	}
-
 	ut_a(node->pcur->rel_pos == BTR_PCUR_ON);
 
 	/* MySQL seems to call rnd_pos before updating each row it
@@ -2029,6 +1992,50 @@ row_update_for_mysql_using_upd_graph(
 	thr->fk_cascade_depth = 0;
 
 run_again:
+	if (DICT_TF2_FLAG_IS_SET(node->table, DICT_TF2_VERSIONED) &&
+		(node->is_delete || node->versioned))
+	{
+		/* System Versioning: modify update vector to set
+		   sys_trx_start (or sys_trx_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 = false;
+			node->vers_delete = true;
+			col_idx = table->vers_row_end;
+		} else {
+			ut_ad(uvect->n_fields < table->n_cols);
+			ufield = &uvect->fields[uvect->n_fields];
+			col_idx = table->vers_row_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;
+
+		static const ulint fsize = sizeof(trx_id_t);
+		byte* buf = static_cast<byte*>(mem_heap_alloc(node->update->heap, fsize));
+		mach_write_to_8(buf, trx->id);
+		dfield_t* dfield = &ufield->new_val;
+		dfield_set_data(dfield, buf, fsize);
+		dict_col_copy_type(col, &dfield->type);
+
+		uvect->n_fields++;
+		ut_ad(node->in_mysql_interface); // otherwise needs to recalculate node->cmpl_info
+	}
+
 	if (thr->fk_cascade_depth == 1 && trx->dict_operation_lock_mode == 0) {
 		got_s_lock = true;
 		row_mysql_freeze_data_dictionary(trx);

storage/innobase/row/row0upd.cc

@@ -269,6 +269,7 @@ row_upd_check_references_constraints(
 
 		if (foreign->referenced_index == index
 		    && (node->is_delete
+			|| node->vers_delete
 			|| row_upd_changes_first_fields_binary(
 				entry, index, node->update,
 				foreign->n_fields))) {
@@ -381,6 +382,7 @@ wsrep_row_upd_check_foreign_constraints(
 
 		if (foreign->foreign_index == index
 		    && (node->is_delete
+			|| node->vers_delete
 			|| row_upd_changes_first_fields_binary(
 				entry, index, node->update,
 				foreign->n_fields))) {