Clean up the logging of virtual column values in table-rebuilding online ALTER

In online table-rebuilding ALTER TABLE (LOCK=NONE), virtual column values
are being written to the online_log. WL#8149 in MySQL 5.7 changed some
low-level functions that are also being used outside row0log.cc, causing
performance penalty to other code.

We revert those changes, and introduce separate functions for writing
the virtual column values.

The only functional change should be the one that is mentioned in
MDEV-13795: row_log_table_low_redundant() will no longer write
virtual column values along with old_pk, just like row_log_table_low().

As noted in MDEV-13795, some forms of table-rebuilding ALTER with
virtual columns is broken. At least DROP PRIMARY KEY, ADD PRIMARY KEY
is broken.

rec_get_converted_size_temp(), rec_convert_dtuple_to_temp():
Remove the parameter for passing virtual column values.

rec_get_converted_size_temp_v(), rec_convert_dtuple_to_temp_v():
New functions for appending virtual column values to the online_log.

rec_get_converted_size_comp_prefix_low(),
rec_convert_dtuple_to_rec_comp(): Remove the v_entry parameter,
and do not allow n_fields=0.

4 files changed, 151 insertions, 132 deletions

diff --git a/storage/innobase/include/rem0rec.h b/storage/innobase/include/rem0rec.h
index a5e3268b7d7..61220d4f533 100644
--- a/storage/innobase/include/rem0rec.h
+++ b/storage/innobase/include/rem0rec.h
@@ -777,19 +777,42 @@ rec_copy(
 	const rec_t*	rec,
 	const ulint*	offsets);
 
-/**********************************************************//**
-Determines the size of a data tuple prefix in a temporary file.
-@return total size */
+/** Determine the size of a data tuple prefix in a temporary file.
+@param[in]	index		clustered or secondary index
+@param[in]	fields		data fields
+@param[in]	n_fields	number of data fields
+@param[out]	extra		record header size
+@return	total size, in bytes */
 ulint
 rec_get_converted_size_temp(
-/*========================*/
-	const dict_index_t*	index,	/*!< in: record descriptor */
-	const dfield_t*		fields,	/*!< in: array of data fields */
-	ulint			n_fields,/*!< in: number of data fields */
-	const dtuple_t*		v_entry,/*!< in: dtuple contains virtual column
-					data */
-	ulint*			extra)	/*!< out: extra size */
-	MY_ATTRIBUTE((warn_unused_result));
+	const dict_index_t*	index,
+	const dfield_t*		fields,
+	ulint			n_fields,
+	ulint*			extra)
+	MY_ATTRIBUTE((warn_unused_result, nonnull(1,2)));
+
+/** Determine the converted size of virtual column data in a temporary file.
+@see rec_convert_dtuple_to_temp_v()
+@param[in]	index	clustered index
+@param[in]	v	clustered index record augmented with the values
+			of virtual columns
+@return size in bytes */
+ulint
+rec_get_converted_size_temp_v(const dict_index_t* index, const dtuple_t* v)
+	MY_ATTRIBUTE((warn_unused_result, nonnull));
+
+/** Write indexed virtual column data into a temporary file.
+@see rec_get_converted_size_temp_v()
+@param[out]	rec	serialized record
+@param[in]	index	clustered index
+@param[in]	v_entry	clustered index record augmented with the values
+			of virtual columns */
+void
+rec_convert_dtuple_to_temp_v(
+	byte*			rec,
+	const dict_index_t*	index,
+	const dtuple_t*		v_entry)
+	MY_ATTRIBUTE((nonnull));
 
 /******************************************************//**
 Determine the offset to each field in temporary file.
@@ -812,10 +835,7 @@ rec_convert_dtuple_to_temp(
 	rec_t*			rec,		/*!< out: record */
 	const dict_index_t*	index,		/*!< in: record descriptor */
 	const dfield_t*		fields,		/*!< in: array of data fields */
-	ulint			n_fields,	/*!< in: number of fields */
-	const dtuple_t*		v_entry);	/*!< in: dtuple contains
-						virtual column data */
-
+	ulint			n_fields);	/*!< in: number of fields */
 
 /**************************************************************//**
 Copies the first n fields of a physical record to a new physical record in
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index 4ad4730f37d..66e8ccec178 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -786,8 +786,6 @@ rec_get_converted_size_comp_prefix_low(
 					it does not */
 	const dfield_t*		fields,	/*!< in: array of data fields */
 	ulint			n_fields,/*!< in: number of data fields */
-	const dtuple_t*		v_entry,/*!< in: dtuple contains virtual column
-					data */
 	ulint*			extra,	/*!< out: extra size */
 	bool			temp)	/*!< in: whether this is a
 					temporary file record */
@@ -795,20 +793,15 @@ rec_get_converted_size_comp_prefix_low(
 	ulint	extra_size;
 	ulint	data_size;
 	ulint	i;
-	ulint	n_null	= (n_fields > 0) ? index->n_nullable : 0;
-	ulint	n_v_fields;
+	ut_ad(n_fields > 0);
 	ut_ad(n_fields <= dict_index_get_n_fields(index));
 	ut_ad(!temp || extra);
 
-	/* At the time being, only temp file record could possible
-	store virtual columns */
-	ut_ad(!v_entry || (dict_index_is_clust(index) && temp));
-	n_v_fields = v_entry ? dtuple_get_n_v_fields(v_entry) : 0;
+	ut_d(ulint n_null = index->n_nullable);
 
 	extra_size = temp
-		? UT_BITS_IN_BYTES(n_null)
-		: REC_N_NEW_EXTRA_BYTES
-		+ UT_BITS_IN_BYTES(n_null);
+		? UT_BITS_IN_BYTES(index->n_nullable)
+		: REC_N_NEW_EXTRA_BYTES + UT_BITS_IN_BYTES(index->n_nullable);
 	data_size = 0;
 
 	if (temp && dict_table_is_comp(index->table)) {
@@ -910,42 +903,50 @@ rec_get_converted_size_comp_prefix_low(
 		*extra = extra_size;
 	}
 
-	/* Log virtual columns */
-	if (n_v_fields != 0) {
-		/* length marker */
-		data_size += 2;
-
-		for (i = 0; i < n_v_fields; i++) {
-			dfield_t*       vfield;
-			ulint		flen;
-
-                        const dict_v_col_t*     col
-                                = dict_table_get_nth_v_col(index->table, i);
-
-			/* Only those indexed needs to be logged */
-                        if (col->m_col.ord_part) {
-				data_size += mach_get_compressed_size(
-					i + REC_MAX_N_FIELDS);
-				vfield = dtuple_get_nth_v_field(
-                                                v_entry, col->v_pos);
-
-                                flen = vfield->len;
-
-				if (flen != UNIV_SQL_NULL) {
-                                        flen = ut_min(
-                                                flen,
-                                                static_cast<ulint>(
-                                                DICT_MAX_FIELD_LEN_BY_FORMAT(
-                                                        index->table)));
-					data_size += flen;
-                                }
-
-				data_size += mach_get_compressed_size(flen);
-			}
+	return(extra_size + data_size);
+}
+
+/** Determine the converted size of virtual column data in a temporary file.
+@see rec_convert_dtuple_to_temp_v()
+@param[in]	index	clustered index
+@param[in]	v	clustered index record augmented with the values
+			of virtual columns
+@return size in bytes */
+ulint
+rec_get_converted_size_temp_v(const dict_index_t* index, const dtuple_t* v)
+{
+	ut_ad(dict_index_is_clust(index));
+
+	/* length marker */
+	ulint data_size = 2;
+	const ulint n_v_fields = dtuple_get_n_v_fields(v);
+
+	for (ulint i = 0; i < n_v_fields; i++) {
+		const dict_v_col_t*     col
+			= dict_table_get_nth_v_col(index->table, i);
+
+		/* Only those indexed needs to be logged */
+		if (!col->m_col.ord_part) {
+			continue;
+		}
+
+		data_size += mach_get_compressed_size(i + REC_MAX_N_FIELDS);
+		const dfield_t* vfield = dtuple_get_nth_v_field(v, col->v_pos);
+		ulint		flen = vfield->len;
+
+		if (flen != UNIV_SQL_NULL) {
+			flen = ut_min(
+				flen,
+				static_cast<ulint>(
+					DICT_MAX_FIELD_LEN_BY_FORMAT(
+						index->table)));
+			data_size += flen;
 		}
+
+		data_size += mach_get_compressed_size(flen);
 	}
 
-	return(extra_size + data_size);
+	return(data_size);
 }
 
 /**********************************************************//**
@@ -961,7 +962,7 @@ rec_get_converted_size_comp_prefix(
 {
 	ut_ad(dict_table_is_comp(index->table));
 	return(rec_get_converted_size_comp_prefix_low(
-		       index, fields, n_fields, NULL, extra, false));
+		       index, fields, n_fields, extra, false));
 }
 
 /**********************************************************//**
@@ -1007,7 +1008,7 @@ rec_get_converted_size_comp(
 	}
 
 	return(size + rec_get_converted_size_comp_prefix_low(
-		       index, fields, n_fields, NULL, extra, false));
+		       index, fields, n_fields, extra, false));
 }
 
 /***********************************************************//**
@@ -1190,8 +1191,6 @@ rec_convert_dtuple_to_rec_comp(
 	const dict_index_t*	index,	/*!< in: record descriptor */
 	const dfield_t*		fields,	/*!< in: array of data fields */
 	ulint			n_fields,/*!< in: number of data fields */
-	const dtuple_t*		v_entry,/*!< in: dtuple contains
-					virtual column data */
 	ulint			status,	/*!< in: status bits of the record */
 	bool			temp)	/*!< in: whether to use the
 					format for temporary files in
@@ -1207,10 +1206,11 @@ rec_convert_dtuple_to_rec_comp(
 	ulint		n_node_ptr_field;
 	ulint		fixed_len;
 	ulint		null_mask	= 1;
-	ulint		n_null;
-	ulint		num_v = v_entry ? dtuple_get_n_v_fields(v_entry) : 0;
 
+	ut_ad(n_fields > 0);
 	ut_ad(temp || dict_table_is_comp(index->table));
+	ulint n_null = index->n_nullable;
+	const ulint n_null_bytes = UT_BITS_IN_BYTES(n_null);
 
 	if (temp) {
 		ut_ad(status == REC_STATUS_ORDINARY);
@@ -1223,8 +1223,6 @@ rec_convert_dtuple_to_rec_comp(
 			temp = false;
 		}
 	} else {
-		ut_ad(v_entry == NULL);
-		ut_ad(num_v == 0);
 		nulls = rec - (REC_N_NEW_EXTRA_BYTES + 1);
 
 		switch (UNIV_EXPECT(status, REC_STATUS_ORDINARY)) {
@@ -1250,13 +1248,9 @@ rec_convert_dtuple_to_rec_comp(
 	}
 
 	end = rec;
-
-	if (n_fields != 0) {
-		n_null = index->n_nullable;
-		lens = nulls - UT_BITS_IN_BYTES(n_null);
-		/* clear the SQL-null flags */
-		memset(lens + 1, 0, nulls - lens);
-	}
+	/* clear the SQL-null flags */
+	lens = nulls - n_null_bytes;
+	memset(lens + 1, 0, nulls - lens);
 
 	/* Store the data and the offsets */
 
@@ -1351,13 +1345,25 @@ rec_convert_dtuple_to_rec_comp(
 			end += len;
 		}
 	}
+}
 
-	if (!num_v) {
-		return;
-	}
+/** Write indexed virtual column data into a temporary file.
+@see rec_get_converted_size_temp_v()
+@param[out]	rec	serialized record
+@param[in]	index	clustered index
+@param[in]	v_entry	clustered index record augmented with the values
+			of virtual columns */
+void
+rec_convert_dtuple_to_temp_v(
+	byte*			rec,
+	const dict_index_t*	index,
+	const dtuple_t*		v_entry)
+{
+	ut_ad(dict_index_is_clust(index));
+	const ulint num_v = dtuple_get_n_v_fields(v_entry);
 
 	/* reserve 2 bytes for writing length */
-	byte*	ptr = end;
+	byte*	ptr = rec;
 	ptr += 2;
 
 	/* Now log information on indexed virtual columns */
@@ -1400,7 +1406,7 @@ rec_convert_dtuple_to_rec_comp(
 		}
 	}
 
-	mach_write_to_2(end, ptr - end);
+	mach_write_to_2(rec, ptr - rec);
 }
 
 /*********************************************************//**
@@ -1426,8 +1432,7 @@ rec_convert_dtuple_to_rec_new(
 	rec = buf + extra_size;
 
 	rec_convert_dtuple_to_rec_comp(
-		rec, index, dtuple->fields, dtuple->n_fields, NULL,
-		status, false);
+		rec, index, dtuple->fields, dtuple->n_fields, status, false);
 
 	/* Set the info bits of the record */
 	rec_set_info_and_status_bits(rec, dtuple_get_info_bits(dtuple));
@@ -1490,21 +1495,21 @@ rec_convert_dtuple_to_rec(
 	return(rec);
 }
 
-/**********************************************************//**
-Determines the size of a data tuple prefix in ROW_FORMAT=COMPACT.
-@return total size */
+/** Determine the size of a data tuple prefix in a temporary file.
+@param[in]	index		clustered or secondary index
+@param[in]	fields		data fields
+@param[in]	n_fields	number of data fields
+@param[out]	extra		record header size
+@return	total size, in bytes */
 ulint
 rec_get_converted_size_temp(
-/*========================*/
-	const dict_index_t*	index,	/*!< in: record descriptor */
-	const dfield_t*		fields,	/*!< in: array of data fields */
-	ulint			n_fields,/*!< in: number of data fields */
-	const dtuple_t*		v_entry,/*!< in: dtuple contains virtual column
-					data */
-	ulint*			extra)	/*!< out: extra size */
+	const dict_index_t*	index,
+	const dfield_t*		fields,
+	ulint			n_fields,
+	ulint*			extra)
 {
 	return(rec_get_converted_size_comp_prefix_low(
-		       index, fields, n_fields, v_entry, extra, true));
+		       index, fields, n_fields, extra, true));
 }
 
 /******************************************************//**
@@ -1530,11 +1535,9 @@ rec_convert_dtuple_to_temp(
 	rec_t*			rec,		/*!< out: record */
 	const dict_index_t*	index,		/*!< in: record descriptor */
 	const dfield_t*		fields,		/*!< in: array of data fields */
-	ulint			n_fields,	/*!< in: number of fields */
-	const dtuple_t*		v_entry)	/*!< in: dtuple contains
-						virtual column data */
+	ulint			n_fields)	/*!< in: number of fields */
 {
-	rec_convert_dtuple_to_rec_comp(rec, index, fields, n_fields, v_entry,
+	rec_convert_dtuple_to_rec_comp(rec, index, fields, n_fields,
 				       REC_STATUS_ORDINARY, true);
 }
 
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index 31a5402dc15..9e946757df3 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -302,7 +302,7 @@ row_log_online_op(
 	extra_size+1 (and reserve 0 as the end-of-chunk marker). */
 
 	size = rec_get_converted_size_temp(
-		index, tuple->fields, tuple->n_fields, NULL, &extra_size);
+		index, tuple->fields, tuple->n_fields, &extra_size);
 	ut_ad(size >= extra_size);
 	ut_ad(size <= sizeof log->tail.buf);
 
@@ -350,7 +350,7 @@ row_log_online_op(
 	}
 
 	rec_convert_dtuple_to_temp(
-		b + extra_size, index, tuple->fields, tuple->n_fields, NULL);
+		b + extra_size, index, tuple->fields, tuple->n_fields);
 	b += size;
 
 	if (mrec_size >= avail_size) {
@@ -641,7 +641,7 @@ row_log_table_delete(
 	ut_ad(DATA_ROLL_PTR_LEN == dtuple_get_nth_field(
 		      old_pk, old_pk->n_fields - 1)->len);
 	old_pk_size = rec_get_converted_size_temp(
-		new_index, old_pk->fields, old_pk->n_fields, NULL,
+		new_index, old_pk->fields, old_pk->n_fields,
 		&old_pk_extra_size);
 	ut_ad(old_pk_extra_size < 0x100);
 
@@ -671,9 +671,7 @@ row_log_table_delete(
 
 	/* Check if we need to log virtual column data */
 	if (ventry->n_v_fields > 0) {
-		ulint	v_extra;
-		mrec_size += rec_get_converted_size_temp(
-			new_index, NULL, 0, ventry, &v_extra);
+		mrec_size += rec_get_converted_size_temp_v(new_index, ventry);
 	}
 
 	if (byte* b = row_log_table_open(index->online_log,
@@ -687,7 +685,7 @@ row_log_table_delete(
 
 		rec_convert_dtuple_to_temp(
 			b + old_pk_extra_size, new_index,
-			old_pk->fields, old_pk->n_fields, NULL);
+			old_pk->fields, old_pk->n_fields);
 
 		b += old_pk_size;
 
@@ -720,8 +718,7 @@ row_log_table_delete(
 
 		/* log virtual columns */
 		if (ventry->n_v_fields > 0) {
-                        rec_convert_dtuple_to_temp(
-                                b, new_index, NULL, 0, ventry);
+                        rec_convert_dtuple_to_temp_v(b, new_index, ventry);
                         b += mach_read_from_2(b);
                 }
 
@@ -812,15 +809,16 @@ row_log_table_low_redundant(
 	}
 
 	size = rec_get_converted_size_temp(
-		index, tuple->fields, tuple->n_fields, ventry, &extra_size);
+		index, tuple->fields, tuple->n_fields, &extra_size);
+	ulint v_size = ventry
+		? rec_get_converted_size_temp_v(index, ventry) : 0;
 
-	mrec_size = ROW_LOG_HEADER_SIZE + size + (extra_size >= 0x80);
+	mrec_size = ROW_LOG_HEADER_SIZE + size + v_size + (extra_size >= 0x80);
 
 	if (num_v) {
 		if (o_ventry) {
-			ulint	v_extra = 0;
-			mrec_size += rec_get_converted_size_temp(
-				index, NULL, 0, o_ventry, &v_extra);
+			mrec_size += rec_get_converted_size_temp_v(
+				index, o_ventry);
 		}
 	} else if (index->table->n_v_cols) {
 		mrec_size += 2;
@@ -839,7 +837,7 @@ row_log_table_low_redundant(
 
 		old_pk_size = rec_get_converted_size_temp(
 			new_index, old_pk->fields, old_pk->n_fields,
-			ventry, &old_pk_extra_size);
+			&old_pk_extra_size);
 		ut_ad(old_pk_extra_size < 0x100);
 		mrec_size += 1/*old_pk_extra_size*/ + old_pk_size;
 	}
@@ -853,8 +851,7 @@ row_log_table_low_redundant(
 
 			rec_convert_dtuple_to_temp(
 				b + old_pk_extra_size, new_index,
-				old_pk->fields, old_pk->n_fields,
-				ventry);
+				old_pk->fields, old_pk->n_fields);
 			b += old_pk_size;
 		}
 
@@ -867,14 +864,17 @@ row_log_table_low_redundant(
 		}
 
 		rec_convert_dtuple_to_temp(
-			b + extra_size, index, tuple->fields, tuple->n_fields,
-			ventry);
+			b + extra_size, index, tuple->fields, tuple->n_fields);
 		b += size;
+		if (ventry) {
+                        rec_convert_dtuple_to_temp_v(b, new_index, ventry);
+			b += v_size;
+		}
 
 		if (num_v) {
 			if (o_ventry) {
-				rec_convert_dtuple_to_temp(
-					b, new_index, NULL, 0, o_ventry);
+				rec_convert_dtuple_to_temp_v(
+					b, new_index, o_ventry);
 				b += mach_read_from_2(b);
 			}
 		} else if (index->table->n_v_cols) {
@@ -964,13 +964,11 @@ row_log_table_low(
 		+ (extra_size >= 0x80) + rec_offs_size(offsets) - omit_size;
 
 	if (ventry && ventry->n_v_fields > 0) {
-		ulint	v_extra = 0;
-		mrec_size += rec_get_converted_size_temp(
-			new_index, NULL, 0, ventry, &v_extra);
+		mrec_size += rec_get_converted_size_temp_v(new_index, ventry);
 
 		if (o_ventry) {
-			mrec_size += rec_get_converted_size_temp(
-				new_index, NULL, 0, o_ventry, &v_extra);
+			mrec_size += rec_get_converted_size_temp_v(
+				new_index, o_ventry);
 		}
 	} else if (index->table->n_v_cols) {
 		/* Always leave 2 bytes length marker for virtual column
@@ -992,7 +990,7 @@ row_log_table_low(
 
 		old_pk_size = rec_get_converted_size_temp(
 			new_index, old_pk->fields, old_pk->n_fields,
-			NULL, &old_pk_extra_size);
+			&old_pk_extra_size);
 		ut_ad(old_pk_extra_size < 0x100);
 		mrec_size += 1/*old_pk_extra_size*/ + old_pk_size;
 	}
@@ -1006,8 +1004,7 @@ row_log_table_low(
 
 			rec_convert_dtuple_to_temp(
 				b + old_pk_extra_size, new_index,
-				old_pk->fields, old_pk->n_fields,
-				NULL);
+				old_pk->fields, old_pk->n_fields);
 			b += old_pk_size;
 		}
 
@@ -1025,13 +1022,12 @@ row_log_table_low(
 		b += rec_offs_data_size(offsets);
 
 		if (ventry && ventry->n_v_fields > 0) {
-			rec_convert_dtuple_to_temp(
-				b, new_index, NULL, 0, ventry);
+			rec_convert_dtuple_to_temp_v(b, new_index, ventry);
 			b += mach_read_from_2(b);
 
 			if (o_ventry) {
-				rec_convert_dtuple_to_temp(
-					b, new_index, NULL, 0, o_ventry);
+				rec_convert_dtuple_to_temp_v(
+					b, new_index, o_ventry);
 				b += mach_read_from_2(b);
 			}
 		} else if (index->table->n_v_cols) {
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index cba453ced24..85fac6c0bba 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -407,7 +407,7 @@ row_merge_buf_encode(
 	ulint	extra_size;
 
 	size = rec_get_converted_size_temp(
-		index, entry->fields, n_fields, NULL, &extra_size);
+		index, entry->fields, n_fields, &extra_size);
 	ut_ad(size >= extra_size);
 
 	/* Encode extra_size + 1 */
@@ -420,7 +420,7 @@ row_merge_buf_encode(
 	}
 
 	rec_convert_dtuple_to_temp(*b + extra_size, index,
-				   entry->fields, n_fields, NULL);
+				   entry->fields, n_fields);
 
 	*b += size;
 }
@@ -893,7 +893,7 @@ row_merge_buf_add(
 		ulint	extra;
 
 		size = rec_get_converted_size_temp(
-			index, entry->fields, n_fields, NULL, &extra);
+			index, entry->fields, n_fields, &extra);
 
 		ut_ad(data_size + extra_size == size);
 		ut_ad(extra_size == extra);