MDEV-21138 Assertion col->ord_part' or f.col->ord_part' failed in row_build_index_entry_low

First part (row0mysql.cc) fixes ins_node_set_new_row() usage workflow
as it is designed to operate on empty row (see row_get_prebuilt_insert_row()
for example).

Second part (row0ins.cc) fixes duplicate key error in FTS_DOC_ID_INDEX
since history rows must not generate entries in that index. We detect
FTS_DOC_ID_INDEX by a number of attributes and skip it if the row is
historical.

Misc fixes:

row_build_index_entry_low() does not accept non-NULL tuple
for FTS index (subject assertion fails), assertion (index->type !=
DICT_FTS) adds code understanding.

Now as historical_row is copied in row_update_vers_insert() there is
no need to copy the row twice: ROW_COPY_POINTERS is used to build
historical_row initially.

dbug_print_rec() debug functions.

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

@@ -130,3 +130,22 @@ ERROR 42S02: Table 'test.xx' doesn't exist
 drop procedure pr;
 drop trigger tr;
 drop table t1;
+#
+# MDEV-21138 Assertion `col->ord_part' or `f.col->ord_part' failed in row_build_index_entry_low
+#
+create table t1 (
+f1 int, f2 text, f3 int, fulltext (f2), key(f1), key(f3),
+foreign key r (f3) references t1 (f1) on delete set null)
+with system versioning engine innodb;
+insert into t1 values (1, repeat('a', 8193), 1), (1, repeat('b', 8193), 1);
+select f1, f3, check_row(row_start, row_end) from t1;
+f1	f3	check_row(row_start, row_end)
+1	1	CURRENT ROW
+1	1	CURRENT ROW
+delete from t1;
+select f1, f3, check_row(row_start, row_end) from t1 for system_time all;
+f1	f3	check_row(row_start, row_end)
+1	1	HISTORICAL ROW
+1	NULL	ERROR: row_end == row_start
+1	1	HISTORICAL ROW
+drop table t1;

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

@@ -241,6 +241,26 @@ B2	salary
 1	2500
 drop table t1;
 drop table t2;
+# Ensure FTS retains correct history
+create table t1 (
+x int, y text, fulltext (y),
+row_start SYS_DATATYPE as row start invisible,
+row_end SYS_DATATYPE as row end invisible,
+period for system_time (row_start, row_end))
+with system versioning engine innodb;
+insert into t1 values (1, repeat('LONG', 2048));
+update t1 set x= x + 1;
+select x, left(y, 4), length(y), check_row(row_start, row_end) from t1 for system_time all order by x, y;
+x	left(y, 4)	length(y)	check_row(row_start, row_end)
+1	LONG	8192	HISTORICAL ROW
+2	LONG	8192	CURRENT ROW
+update t1 set y= 'SHORT';
+select x, left(y, 4), length(y), check_row(row_start, row_end) from t1 for system_time all order by x, y;
+x	left(y, 4)	length(y)	check_row(row_start, row_end)
+1	LONG	8192	HISTORICAL ROW
+2	LONG	8192	HISTORICAL ROW
+2	SHOR	5	CURRENT ROW
+drop tables t1;
 ### Issue tempesta-tech/mariadb#365, bug 7 (duplicate of historical row)
 create or replace table t1 (a int primary key, b int)
 with system versioning engine myisam;

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

@@ -94,4 +94,19 @@ drop procedure pr;
 drop trigger tr;
 drop table t1;
 
+--echo #
+--echo # MDEV-21138 Assertion `col->ord_part' or `f.col->ord_part' failed in row_build_index_entry_low
+--echo #
+create table t1 (
+  f1 int, f2 text, f3 int, fulltext (f2), key(f1), key(f3),
+  foreign key r (f3) references t1 (f1) on delete set null)
+with system versioning engine innodb;
+insert into t1 values (1, repeat('a', 8193), 1), (1, repeat('b', 8193), 1);
+select f1, f3, check_row(row_start, row_end) from t1;
+delete from t1;
+select f1, f3, check_row(row_start, row_end) from t1 for system_time all;
+
+# cleanup
+drop table t1;
+
 --source suite/versioning/common_finish.inc

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

@@ -147,6 +147,21 @@ select @tmp2 = sys_trx_start as B2, salary from t2;
 drop table t1;
 drop table t2;
 
+--echo # Ensure FTS retains correct history
+replace_result $sys_datatype_expl SYS_DATATYPE;
+eval create table t1 (
+  x int, y text, fulltext (y),
+  row_start $sys_datatype_expl as row start invisible,
+  row_end $sys_datatype_expl as row end invisible,
+  period for system_time (row_start, row_end))
+with system versioning engine innodb;
+insert into t1 values (1, repeat('LONG', 2048));
+update t1 set x= x + 1;
+select x, left(y, 4), length(y), check_row(row_start, row_end) from t1 for system_time all order by x, y;
+update t1 set y= 'SHORT';
+select x, left(y, 4), length(y), check_row(row_start, row_end) from t1 for system_time all order by x, y;
+drop tables t1;
+
 --echo ### Issue tempesta-tech/mariadb#365, bug 7 (duplicate of historical row)
 create or replace table t1 (a int primary key, b int)
 with system versioning engine myisam;

storage/innobase/include/row0ins.h

@@ -206,6 +206,7 @@ struct ins_node_t
 				if this is NULL, entry list should be created
 				and buffers for sys fields in row allocated */
         void vers_update_end(row_prebuilt_t *prebuilt, bool history_row);
+	bool vers_history_row() const; /* true if 'row' is historical */
 };
 
 /** Create an insert object.

storage/innobase/row/row0ins.cc

@@ -3566,6 +3566,16 @@ row_ins_get_row_from_select(
 	}
 }
 
+inline
+bool ins_node_t::vers_history_row() const
+{
+	if (!table->versioned())
+		return false;
+	dfield_t* row_end = dtuple_get_nth_field(row, table->vers_end);
+	return row_end->vers_history_row();
+}
+
+
 /***********************************************************//**
 Inserts a row to a table.
 @return DB_SUCCESS if operation successfully completed, else error
@@ -3604,12 +3614,31 @@ row_ins(
 	ut_ad(node->state == INS_NODE_INSERT_ENTRIES);
 
 	while (node->index != NULL) {
-		if (node->index->type != DICT_FTS) {
+		dict_index_t *index = node->index;
+		/*
+		   We do not insert history rows into FTS_DOC_ID_INDEX because
+		   it is unique by FTS_DOC_ID only and we do not want to add
+		   row_end to unique key. Fulltext field works the way new
+		   FTS_DOC_ID is created on every fulltext UPDATE, so holding only
+		   FTS_DOC_ID for history is enough.
+		*/
+		const unsigned type = index->type;
+		if (index->type & DICT_FTS) {
+		} else if (!(type & DICT_UNIQUE) || index->n_uniq > 1
+			   || !node->vers_history_row()) {
+
 			dberr_t err = row_ins_index_entry_step(node, thr);
 
 			if (err != DB_SUCCESS) {
 				DBUG_RETURN(err);
 			}
+		} else {
+			/* Unique indexes with system versioning must contain
+			the version end column. The only exception is a hidden
+			FTS_DOC_ID_INDEX that InnoDB may create on a hidden or
+			user-created FTS_DOC_ID column. */
+			ut_ad(!strcmp(index->name, FTS_DOC_ID_INDEX_NAME));
+			ut_ad(!strcmp(index->fields[0].name, FTS_DOC_ID_COL_NAME));
 		}
 
 		node->index = dict_table_get_next_index(node->index);

storage/innobase/row/row0mysql.cc

@@ -2104,10 +2104,18 @@ row_mysql_unfreeze_data_dictionary(
 @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.
+/** Insert history row when evaluating foreign key referential action.
+
+1. Create new dtuple_t 'row' from node->historical_row;
+2. Update its row_end to current timestamp;
+3. Insert it to a table;
+4. Update table statistics.
+
+This is used in UPDATE CASCADE/SET NULL of a system versioned referenced table.
+
+node->historical_row: dtuple_t containing pointers of row changed by refertial
+action.
+
 @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 */
@@ -2119,9 +2127,16 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 	dict_table_t* table = node->table;
 	ut_ad(table->versioned());
 
-	dtuple_t* row = node->historical_row;
-	ut_ad(row);
-	node->historical_row = NULL;
+	dtuple_t*       row;
+	const ulint     n_cols        = dict_table_get_n_cols(table);
+	const ulint     n_v_cols      = dict_table_get_n_v_cols(table);
+
+	ut_ad(n_cols == dtuple_get_n_fields(node->historical_row));
+	ut_ad(n_v_cols == dtuple_get_n_v_fields(node->historical_row));
+
+	row = dtuple_create_with_vcol(node->historical_heap, n_cols, n_v_cols);
+
+	dict_table_copy_types(row, table);
 
 	ins_node_t* insert_node =
 		ins_node_create(INS_DIRECT, table, node->historical_heap);
@@ -2134,6 +2149,22 @@ static dberr_t row_update_vers_insert(que_thr_t* thr, upd_node_t* node)
 	insert_node->common.parent = thr;
 	ins_node_set_new_row(insert_node, row);
 
+	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_copy(dst, src);
+	}
+
+	for (ulint i = 0; i < n_v_cols; i++) {
+		dfield_t *dst= dtuple_get_nth_v_field(row, i);
+		dfield_t *src= dtuple_get_nth_v_field(node->historical_row, i);
+		dfield_copy(dst, src);
+	}
+
+	node->historical_row = NULL;
+
 	row_end = dtuple_get_nth_field(row, table->vers_end);
 	if (dict_table_get_nth_col(table, table->vers_end)->vers_native()) {
 		mach_write_to_8(row_end_data, trx->id);

storage/innobase/row/row0row.cc

@@ -294,6 +294,8 @@ row_build_index_entry_low(
 			continue;
 		}
 
+		ut_ad(!(index->type & DICT_FTS));
+
 		if ((!ind_field || ind_field->prefix_len == 0)
 		    && (!dfield_is_ext(dfield)
 			|| dict_index_is_clust(index))) {

storage/innobase/ut/ut0ut.cc

@@ -38,6 +38,7 @@ Created 5/11/1994 Heikki Tuuri
 #include <string>
 #include "log.h"
 #include "my_cpu.h"
+#include "rem0rec.h"
 
 /**********************************************************//**
 Returns the number of milliseconds since some epoch.  The
@@ -627,4 +628,42 @@ fatal_or_error::~fatal_or_error()
 
 } // namespace ib
 
+#ifndef DBUG_OFF
+const char * dbug_print_rec(const rec_t* rec, const rec_offs* offsets)
+{
+	rec_printer r(rec, offsets);
+	return r.str().c_str();
+}
+
+const char * dbug_print_rec(const rec_t* rec, ulint info, const rec_offs* offsets)
+{
+	rec_printer r(rec, info, offsets);
+	return r.str().c_str();
+}
+
+const char * dbug_print_rec(const dtuple_t* tuple)
+{
+	rec_printer r(tuple);
+	return r.str().c_str();
+}
+
+const char * dbug_print_rec(const dfield_t* field, ulint n)
+{
+	rec_printer r(field, n);
+	return r.str().c_str();
+}
+
+const char * dbug_print_rec(const rec_t* rec, dict_index_t* index)
+{
+	rec_offs	offsets_[REC_OFFS_NORMAL_SIZE];
+	rec_offs*	offsets		= offsets_;
+	rec_offs_init(offsets_);
+	mem_heap_t*	tmp_heap	= NULL;
+	offsets = rec_get_offsets(rec, index, offsets, true,
+				  ULINT_UNDEFINED, &tmp_heap);
+	rec_printer r(rec, offsets);
+	return r.str().c_str();
+}
+#endif /* !DBUG_OFF */
+
 #endif /* !UNIV_INNOCHECKSUM */