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().

3 files changed, 47 insertions, 34 deletions

diff --git a/storage/innobase/include/row0upd.h b/storage/innobase/include/row0upd.h
index b34acfd8dc1..de97b379d9e 100644
--- a/storage/innobase/include/row0upd.h
+++ b/storage/innobase/include/row0upd.h
@@ -587,25 +587,25 @@ private:
 	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
diff --git a/storage/innobase/row/row0mysql.cc b/storage/innobase/row/row0mysql.cc
index b2256e9905d..d2fc63a5f38 100644
--- a/storage/innobase/row/row0mysql.cc
+++ b/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 (;;) {
diff --git a/storage/innobase/row/row0upd.cc b/storage/innobase/row/row0upd.cc
index 0f700c77c36..8ce47e74a1c 100644
--- a/storage/innobase/row/row0upd.cc
+++ b/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);
 }