MDEV-23644 Assertion on evaluating foreign referential action for self-reference in system versioned table

First part of the fix (row0mysql.cc) addresses external columns when adding history
row on referential action. The full data must be retrieved before the
row is inserted.

Second part of the fix (the rest) avoids duplicate primary key error between
the history row generated on referential action and the history row
generated by SQL command. Both command and referential action can
happen on same table since foreign key can be self-reference (parent
and child tables are same). Moreover, the self-reference can refer
multiple rows when the key is non-unique. In such case history is
generated by referential action occured on first row but processed all
rows by a matched key. The second round is when the next row is
processed by a command but history already exists. In such case we
check TRX_ID of existing history row and if it is the same we assume
the above situation and skip adding one more history row or failing
the command.

4 files changed, 43 insertions, 2 deletions

diff --git a/storage/innobase/data/data0data.cc b/storage/innobase/data/data0data.cc
index 47e58f1614a..3e23cd6f662 100644
--- a/storage/innobase/data/data0data.cc
+++ b/storage/innobase/data/data0data.cc
@@ -738,7 +738,7 @@ void
 dtuple_convert_back_big_rec(
 /*========================*/
 	dict_index_t*	index MY_ATTRIBUTE((unused)),	/*!< in: index */
-	dtuple_t*	entry,	/*!< in: entry whose data was put to vector */
+	dtuple_t*	entry,	/*!< in/out: entry whose data was put to vector */
 	big_rec_t*	vector)	/*!< in, own: big rec vector; it is
 				freed in this function */
 {
diff --git a/storage/innobase/include/data0data.h b/storage/innobase/include/data0data.h
index 11a7f2e516f..002332852b8 100644
--- a/storage/innobase/include/data0data.h
+++ b/storage/innobase/include/data0data.h
@@ -543,6 +543,16 @@ struct dtuple_t {
 	inserted or updated.
 	@param[in]	index	index possibly with instantly added columns */
 	void trim(const dict_index_t& index);
+	bool vers_history_row() const
+	{
+		for (ulint i = 0; i < n_fields; i++) {
+			const dfield_t* field = &fields[i];
+			if (field->type.vers_sys_end()) {
+				return field->vers_history_row();
+			}
+		}
+		return false;
+	}
 };
 
 inline ulint dtuple_get_n_fields(const dtuple_t* tuple)
diff --git a/storage/innobase/row/row0ins.cc b/storage/innobase/row/row0ins.cc
index 88152841293..43f80adebbd 100644
--- a/storage/innobase/row/row0ins.cc
+++ b/storage/innobase/row/row0ins.cc
@@ -2385,6 +2385,18 @@ row_ins_duplicate_error_in_clust(
 duplicate:
 				trx->error_info = cursor->index;
 				err = DB_DUPLICATE_KEY;
+				if (cursor->index->table->versioned()
+				    && entry->vers_history_row())
+				{
+					ulint trx_id_len;
+					byte *trx_id = rec_get_nth_field(
+						rec, offsets, n_unique,
+						&trx_id_len);
+					ut_ad(trx_id_len == DATA_TRX_ID_LEN);
+					if (trx->id == trx_read_trx_id(trx_id)) {
+						err = DB_FOREIGN_DUPLICATE_KEY;
+					}
+				}
 				goto func_exit;
 			}
 		}
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index 848f60c47d0..c643a907722 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/storage/innobase/row/row0mysql.cc
@@ -2125,6 +2125,7 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 	dfield_t* row_end;
 	char row_end_data[8];
 	dict_table_t* table = node->table;
+	page_size_t page_size= dict_table_page_size(table);
 	ut_ad(table->versioned());
 
 	dtuple_t*       row;
@@ -2152,9 +2153,27 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 	ut_ad(n_cols > DATA_N_SYS_COLS);
 	// Exclude DB_ROW_ID, DB_TRX_ID, DB_ROLL_PTR
 	for (ulint i = 0; i < n_cols - DATA_N_SYS_COLS; i++) {
-		dfield_t *dst= dtuple_get_nth_field(row, i);
 		dfield_t *src= dtuple_get_nth_field(node->historical_row, i);
+		dfield_t *dst= dtuple_get_nth_field(row, i);
 		dfield_copy(dst, src);
+		if (dfield_is_ext(src)) {
+			byte *field_data
+				= static_cast<byte*>(dfield_get_data(src));
+			ulint ext_len;
+			ulint field_len = dfield_get_len(src);
+
+			ut_a(field_len >= BTR_EXTERN_FIELD_REF_SIZE);
+
+			ut_a(memcmp(field_data + field_len
+				     - BTR_EXTERN_FIELD_REF_SIZE,
+				     field_ref_zero,
+				     BTR_EXTERN_FIELD_REF_SIZE));
+
+			byte *data = btr_copy_externally_stored_field(
+				&ext_len, field_data, page_size, field_len,
+				node->historical_heap);
+			dfield_set_data(dst, data, ext_len);
+		}
 	}
 
 	for (ulint i = 0; i < n_v_cols; i++) {