MDEV-22061 InnoDB: Assertion of missing row in sec index row_start upon...

MDEV-22061 InnoDB: Assertion of missing row in sec index row_start upon REPLACE on a system-versioned table

make_versioned_helper() appended new update field unconditionally
while it should check if this field already exists in update vector.

Misc renames to conform versioning prefix. vers_update_fields() name
conforms with sql layer TABLE::vers_update_fields().

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

@@ -319,3 +319,19 @@ create or replace table t1 (f point, key(f)) with system versioning engine=myisa
 update t1 set f = null where f = 'foo';
 ERROR 22003: Cannot get geometry object from data you send to the GEOMETRY field
 drop table t1;
+#
+# MDEV-22061 InnoDB: Assertion of missing row in sec index row_start upon REPLACE on a system-versioned table
+#
+create or replace table t1 (
+a int,
+b int,
+row_start bigint(20) unsigned generated always as row start,
+row_end bigint(20) unsigned generated always as row end,
+unique key (b,row_end),
+key (row_start),
+period for system_time (row_start,row_end)
+) engine=innodb with system versioning;
+insert into t1 (a, b) values (1, 2);
+replace into t1 (a, b) values (3, 2);
+replace into t1 (a, b) values (4, 2);
+drop table t1;

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

@@ -245,4 +245,24 @@ update t1 set f = null where f = 'foo';
 # cleanup
 drop table t1;
 
+--echo #
+--echo # MDEV-22061 InnoDB: Assertion of missing row in sec index row_start upon REPLACE on a system-versioned table
+--echo #
+create or replace table t1 (
+  a int,
+  b int,
+  row_start bigint(20) unsigned generated always as row start,
+  row_end bigint(20) unsigned generated always as row end,
+  unique key (b,row_end),
+  key (row_start),
+  period for system_time (row_start,row_end)
+) engine=innodb with system versioning;
+
+insert into t1 (a, b) values (1, 2);
+replace into t1 (a, b) values (3, 2);
+replace into t1 (a, b) values (4, 2);
+
+# cleanup
+drop table t1;
+
 source suite/versioning/common_finish.inc;

storage/innobase/include/row0upd.h

@@ -587,25 +587,25 @@ struct upd_node_t{
 	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);
+  void vers_update_fields(const trx_t *trx, ulint idx);
 
 public:
 	/** Also set row_start = CURRENT_TIMESTAMP/trx->id
 	@param[in]	trx	transaction */
-	void make_versioned_update(const trx_t* trx)
-	{
-		make_versioned_helper(trx, table->vers_start);
-	}
+  void vers_make_update(const trx_t *trx)
+  {
+    vers_update_fields(trx, table->vers_start);
+        }
 
 	/** 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)
-	{
+        void vers_make_delete(const trx_t *trx)
+        {
 		update->n_fields = 0;
 		is_delete = VERSIONED_DELETE;
-		make_versioned_helper(trx, table->vers_end);
-	}
+                vers_update_fields(trx, table->vers_end);
+        }
 };
 
 #define	UPD_NODE_MAGIC_N	1579975

storage/innobase/row/row0mysql.cc

@@ -1871,10 +1871,10 @@ row_update_for_mysql(row_prebuilt_t* prebuilt)
 
 	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);
-		}
+                  node->vers_make_delete(trx);
+                } else if (node->update->affects_versioned()) {
+                  node->vers_make_update(trx);
+                }
 	}
 
 	for (;;) {
@@ -2239,14 +2239,14 @@ row_update_cascade_for_mysql(
 
 	if (table->versioned()) {
 		if (node->is_delete == PLAIN_DELETE) {
-			node->make_versioned_delete(trx);
-		} else if (node->update->affects_versioned()) {
+                  node->vers_make_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);
-		}
+                        node->vers_make_update(trx);
+                }
 	}
 
 	for (;;) {

storage/innobase/row/row0upd.cc

@@ -3486,32 +3486,45 @@ 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)
+void upd_node_t::vers_update_fields(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);
-
-	/* row_create_update_node_for_mysql() pre-allocated this much.
+        const dict_col_t *col= dict_table_get_nth_col(table, idx);
+        ulint field_no= dict_col_get_clust_pos(col, clust_index);
+        upd_field_t *ufield;
+
+        for (ulint i= 0; i < update->n_fields; ++i)
+        {
+          if (update->fields[i].field_no == field_no)
+          {
+            ufield= &update->fields[i];
+            goto skip_append;
+          }
+        }
+
+        /* row_create_update_node_for_mysql() pre-allocated this much.
 	   At least one PK column always remains unchanged. */
 	ut_ad(update->n_fields < ulint(table->n_cols + table->n_v_cols));
 
 	update->n_fields++;
-	upd_field_t* ufield = upd_get_nth_field(update, update->n_fields - 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);
+        ufield= upd_get_nth_field(update, update->n_fields - 1);
+        upd_field_set_field_no(ufield, field_no, clust_index);
+
+skip_append:
+  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);
 }