Merge 10.3 into 10.4

5 files changed, 78 insertions, 41 deletions

diff --git a/storage/innobase/include/rem0rec.h b/storage/innobase/include/rem0rec.h
index 6f6535c529f..027466e11f4 100644
--- a/storage/innobase/include/rem0rec.h
+++ b/storage/innobase/include/rem0rec.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1994, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2020, MariaDB Corporation.
+Copyright (c) 2017, 2021, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -1034,12 +1034,14 @@ rec_copy(
 	const rec_offs*	offsets);
 
 /** Determine the size of a data tuple prefix in a temporary file.
+@tparam redundant_temp whether to use the ROW_FORMAT=REDUNDANT format
 @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
 @param[in]	status		REC_STATUS_ORDINARY or REC_STATUS_INSTANT
 @return	total size, in bytes */
+template<bool redundant_temp>
 ulint
 rec_get_converted_size_temp(
 	const dict_index_t*	index,
@@ -1078,11 +1080,13 @@ rec_init_offsets_temp(
 	MY_ATTRIBUTE((nonnull));
 
 /** Convert a data tuple prefix to the temporary file format.
+@tparam redundant_temp whether to use the ROW_FORMAT=REDUNDANT format
 @param[out]	rec		record in temporary file format
 @param[in]	index		clustered or secondary index
 @param[in]	fields		data fields
 @param[in]	n_fields	number of data fields
 @param[in]	status		REC_STATUS_ORDINARY or REC_STATUS_INSTANT */
+template<bool redundant_temp>
 void
 rec_convert_dtuple_to_temp(
 	rec_t*			rec,
diff --git a/storage/innobase/include/row0log.h b/storage/innobase/include/row0log.h
index 5ec4b9c1103..93aa5c24230 100644
--- a/storage/innobase/include/row0log.h
+++ b/storage/innobase/include/row0log.h
@@ -247,6 +247,11 @@ row_log_apply(
 	ut_stage_alter_t*	stage)
 	MY_ATTRIBUTE((warn_unused_result));
 
+/** Get the n_core_fields of online log for the index
+@param	 index	index whose n_core_fields of log to be accessed
+@return number of n_core_fields */
+unsigned row_log_get_n_core_fields(const dict_index_t *index);
+
 #ifdef HAVE_PSI_STAGE_INTERFACE
 /** Estimate how much work is to be done by the log apply phase
 of an ALTER TABLE for this index.
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index 1cc11415615..178290190b9 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -29,6 +29,7 @@ Created 5/30/1994 Heikki Tuuri
 #include "mtr0log.h"
 #include "fts0fts.h"
 #include "trx0sys.h"
+#include "row0log.h"
 
 /*			PHYSICAL RECORD (OLD STYLE)
 			===========================
@@ -1094,7 +1095,8 @@ rec_get_nth_field_offs_old(
 }
 
 /** Determine the size of a data tuple prefix in ROW_FORMAT=COMPACT.
-@tparam	mblob	whether the record includes a metadata BLOB
+@tparam	mblob	        whether the record includes a metadata BLOB
+@tparam redundant_temp  whether to use the ROW_FORMAT=REDUNDANT format
 @param[in]	index		record descriptor; dict_table_is_comp()
 				is assumed to hold, even if it doesn't
 @param[in]	dfield		array of data fields
@@ -1103,7 +1105,7 @@ rec_get_nth_field_offs_old(
 @param[in]	status		status flags
 @param[in]	temp		whether this is a temporary file record
 @return total size */
-template<bool mblob = false>
+template<bool mblob = false, bool redundant_temp = false>
 static inline
 ulint
 rec_get_converted_size_comp_prefix_low(
@@ -1120,25 +1122,27 @@ rec_get_converted_size_comp_prefix_low(
 	ut_d(ulint n_null = index->n_nullable);
 	ut_ad(status == REC_STATUS_ORDINARY || status == REC_STATUS_NODE_PTR
 	      || status == REC_STATUS_INSTANT);
+	unsigned n_core_fields = redundant_temp
+		? row_log_get_n_core_fields(index)
+		: index->n_core_fields;
 
 	if (mblob) {
-		ut_ad(!temp);
 		ut_ad(index->table->instant);
-		ut_ad(index->is_instant());
+		ut_ad(!redundant_temp && index->is_instant());
 		ut_ad(status == REC_STATUS_INSTANT);
 		ut_ad(n_fields == ulint(index->n_fields) + 1);
 		extra_size += UT_BITS_IN_BYTES(index->n_nullable)
 			+ rec_get_n_add_field_len(n_fields - 1
-						  - index->n_core_fields);
+						  - n_core_fields);
 	} else if (status == REC_STATUS_INSTANT
-		   && (!temp || n_fields > index->n_core_fields)) {
-		ut_ad(index->is_instant());
+		   && (!temp || n_fields > n_core_fields)) {
+		if (!redundant_temp) { ut_ad(index->is_instant()); }
 		ut_ad(UT_BITS_IN_BYTES(n_null) >= index->n_core_null_bytes);
 		extra_size += UT_BITS_IN_BYTES(index->get_n_nullable(n_fields))
 			+ rec_get_n_add_field_len(n_fields - 1
-						  - index->n_core_fields);
+						  - n_core_fields);
 	} else {
-		ut_ad(n_fields <= index->n_core_fields);
+		ut_ad(n_fields <= n_core_fields);
 		extra_size += index->n_core_null_bytes;
 	}
 
@@ -1489,7 +1493,8 @@ rec_convert_dtuple_to_rec_old(
 }
 
 /** Convert a data tuple into a ROW_FORMAT=COMPACT record.
-@tparam	mblob	whether the record includes a metadata BLOB
+@tparam	mblob	        whether the record includes a metadata BLOB
+@tparam redundant_temp  whether to use the ROW_FORMAT=REDUNDANT format
 @param[out]	rec		converted record
 @param[in]	index		index
 @param[in]	field		data fields to convert
@@ -1497,7 +1502,7 @@ rec_convert_dtuple_to_rec_old(
 @param[in]	status		rec_get_status(rec)
 @param[in]	temp		whether to use the format for temporary files
 				in index creation */
-template<bool mblob = false>
+template<bool mblob = false, bool redundant_temp = false>
 static inline
 void
 rec_convert_dtuple_to_rec_comp(
@@ -1514,7 +1519,9 @@ rec_convert_dtuple_to_rec_comp(
 	byte*		UNINIT_VAR(lens);
 	ulint		UNINIT_VAR(n_node_ptr_field);
 	ulint		null_mask	= 1;
-
+	const ulint	n_core_fields = redundant_temp
+			? row_log_get_n_core_fields(index)
+			: index->n_core_fields;
 	ut_ad(n_fields > 0);
 	ut_ad(temp || dict_table_is_comp(index->table));
 	ut_ad(index->n_core_null_bytes <= UT_BITS_IN_BYTES(index->n_nullable));
@@ -1524,11 +1531,10 @@ rec_convert_dtuple_to_rec_comp(
 	if (mblob) {
 		ut_ad(!temp);
 		ut_ad(index->table->instant);
-		ut_ad(index->is_instant());
+		ut_ad(!redundant_temp && index->is_instant());
 		ut_ad(status == REC_STATUS_INSTANT);
 		ut_ad(n_fields == ulint(index->n_fields) + 1);
-		rec_set_n_add_field(nulls, n_fields - 1
-				    - index->n_core_fields);
+		rec_set_n_add_field(nulls, n_fields - 1 - n_core_fields);
 		rec_set_heap_no_new(rec, PAGE_HEAP_NO_USER_LOW);
 		rec_set_status(rec, REC_STATUS_INSTANT);
 		n_node_ptr_field = ULINT_UNDEFINED;
@@ -1537,20 +1543,17 @@ rec_convert_dtuple_to_rec_comp(
 	}
 	switch (status) {
 	case REC_STATUS_INSTANT:
-		ut_ad(index->is_instant());
-		ut_ad(n_fields > index->n_core_fields);
-		rec_set_n_add_field(nulls, n_fields - 1
-				    - index->n_core_fields);
+		if (!redundant_temp) { ut_ad(index->is_instant()); }
+		ut_ad(n_fields > n_core_fields);
+		rec_set_n_add_field(nulls, n_fields - 1 - n_core_fields);
 		/* fall through */
 	case REC_STATUS_ORDINARY:
 		ut_ad(n_fields <= dict_index_get_n_fields(index));
 		if (!temp) {
 			rec_set_heap_no_new(rec, PAGE_HEAP_NO_USER_LOW);
-
-			rec_set_status(
-				rec, n_fields == index->n_core_fields
-				     ? REC_STATUS_ORDINARY
-				     : REC_STATUS_INSTANT);
+			rec_set_status(rec, n_fields == n_core_fields
+				       ? REC_STATUS_ORDINARY
+				       : REC_STATUS_INSTANT);
 		}
 
 		if (dict_table_is_comp(index->table)) {
@@ -1768,12 +1771,14 @@ rec_convert_dtuple_to_rec(
 }
 
 /** Determine the size of a data tuple prefix in a temporary file.
+@tparam redundant_temp whether to use the ROW_FORMAT=REDUNDANT format
 @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
 @param[in]	status		REC_STATUS_ORDINARY or REC_STATUS_INSTANT
 @return	total size, in bytes */
+template<bool redundant_temp>
 ulint
 rec_get_converted_size_temp(
 	const dict_index_t*	index,
@@ -1782,10 +1787,18 @@ rec_get_converted_size_temp(
 	ulint*			extra,
 	rec_comp_status_t	status)
 {
-	return rec_get_converted_size_comp_prefix_low(
+	return rec_get_converted_size_comp_prefix_low<false,redundant_temp>(
 		index, fields, n_fields, extra, status, true);
 }
 
+template ulint rec_get_converted_size_temp<false>(
+	const dict_index_t*, const dfield_t*, ulint, ulint*,
+	rec_comp_status_t);
+
+template ulint rec_get_converted_size_temp<true>(
+	const dict_index_t*, const dfield_t*, ulint, ulint*,
+	rec_comp_status_t);
+
 /** Determine the offset to each field in temporary file.
 @param[in]	rec	temporary file record
 @param[in]	index	index of that the record belongs to
@@ -1838,6 +1851,7 @@ rec_init_offsets_temp(
 @param[in]	n_fields	number of data fields
 @param[in]	status		REC_STATUS_ORDINARY or REC_STATUS_INSTANT
 */
+template<bool redundant_temp>
 void
 rec_convert_dtuple_to_temp(
 	rec_t*			rec,
@@ -1846,10 +1860,18 @@ rec_convert_dtuple_to_temp(
 	ulint			n_fields,
 	rec_comp_status_t	status)
 {
-	rec_convert_dtuple_to_rec_comp(rec, index, fields, n_fields,
-				       status, true);
+	rec_convert_dtuple_to_rec_comp<false,redundant_temp>(
+		rec, index, fields, n_fields, status, true);
 }
 
+template void rec_convert_dtuple_to_temp<false>(
+	rec_t*, const dict_index_t*, const dfield_t*,
+	ulint, rec_comp_status_t);
+
+template void rec_convert_dtuple_to_temp<true>(
+	rec_t*, const dict_index_t*, const dfield_t*,
+	ulint, rec_comp_status_t);
+
 /** Copy the first n fields of a (copy of a) physical record to a data tuple.
 The fields are copied into the memory heap.
 @param[out]	tuple		data tuple
diff --git a/storage/innobase/row/row0log.cc b/storage/innobase/row/row0log.cc
index 45ec027beb5..6d5f61366fb 100644
--- a/storage/innobase/row/row0log.cc
+++ b/storage/innobase/row/row0log.cc
@@ -353,7 +353,7 @@ row_log_online_op(
 	row_merge_buf_encode(), because here we do not encode
 	extra_size+1 (and reserve 0 as the end-of-chunk marker). */
 
-	size = rec_get_converted_size_temp(
+	size = rec_get_converted_size_temp<false>(
 		index, tuple->fields, tuple->n_fields, &extra_size);
 	ut_ad(size >= extra_size);
 	ut_ad(size <= sizeof log->tail.buf);
@@ -401,7 +401,7 @@ row_log_online_op(
 		*b++ = (byte) extra_size;
 	}
 
-	rec_convert_dtuple_to_temp(
+	rec_convert_dtuple_to_temp<false>(
 		b + extra_size, index, tuple->fields, tuple->n_fields);
 	b += size;
 
@@ -743,7 +743,7 @@ row_log_table_delete(
 		      old_pk, old_pk->n_fields - 2)->len);
 	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(
+	old_pk_size = rec_get_converted_size_temp<false>(
 		new_index, old_pk->fields, old_pk->n_fields,
 		&old_pk_extra_size);
 	ut_ad(old_pk_extra_size < 0x100);
@@ -756,7 +756,7 @@ row_log_table_delete(
 		*b++ = ROW_T_DELETE;
 		*b++ = static_cast<byte>(old_pk_extra_size);
 
-		rec_convert_dtuple_to_temp(
+		rec_convert_dtuple_to_temp<false>(
 			b + old_pk_extra_size, new_index,
 			old_pk->fields, old_pk->n_fields);
 
@@ -856,7 +856,7 @@ row_log_table_low_redundant(
 	rec_comp_status_t status = is_instant
 		? REC_STATUS_INSTANT : REC_STATUS_ORDINARY;
 
-	size = rec_get_converted_size_temp(
+	size = rec_get_converted_size_temp<true>(
 		index, tuple->fields, tuple->n_fields, &extra_size, status);
 	if (is_instant) {
 		size++;
@@ -876,7 +876,7 @@ row_log_table_low_redundant(
 		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(
+		old_pk_size = rec_get_converted_size_temp<false>(
 			new_index, old_pk->fields, old_pk->n_fields,
 			&old_pk_extra_size);
 		ut_ad(old_pk_extra_size < 0x100);
@@ -893,7 +893,7 @@ row_log_table_low_redundant(
 			if (old_pk_size) {
 				*b++ = static_cast<byte>(old_pk_extra_size);
 
-				rec_convert_dtuple_to_temp(
+				rec_convert_dtuple_to_temp<false>(
 					b + old_pk_extra_size, new_index,
 					old_pk->fields, old_pk->n_fields);
 				b += old_pk_size;
@@ -916,7 +916,7 @@ row_log_table_low_redundant(
 			*b = status;
 		}
 
-		rec_convert_dtuple_to_temp(
+		rec_convert_dtuple_to_temp<true>(
 			b + extra_size, index, tuple->fields, tuple->n_fields,
 			status);
 		b += size;
@@ -1038,7 +1038,7 @@ row_log_table_low(
 		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(
+		old_pk_size = rec_get_converted_size_temp<false>(
 			new_index, old_pk->fields, old_pk->n_fields,
 			&old_pk_extra_size);
 		ut_ad(old_pk_extra_size < 0x100);
@@ -1054,7 +1054,7 @@ row_log_table_low(
 			if (old_pk_size) {
 				*b++ = static_cast<byte>(old_pk_extra_size);
 
-				rec_convert_dtuple_to_temp(
+				rec_convert_dtuple_to_temp<false>(
 					b + old_pk_extra_size, new_index,
 					old_pk->fields, old_pk->n_fields);
 				b += old_pk_size;
@@ -4045,3 +4045,9 @@ row_log_apply(
 
 	DBUG_RETURN(error);
 }
+
+unsigned row_log_get_n_core_fields(const dict_index_t *index)
+{
+  ut_ad(index->online_log);
+  return index->online_log->n_core_fields;
+}
diff --git a/storage/innobase/row/row0merge.cc b/storage/innobase/row/row0merge.cc
index 3939d48ea9a..a0dd06f3ddf 100644
--- a/storage/innobase/row/row0merge.cc
+++ b/storage/innobase/row/row0merge.cc
@@ -308,7 +308,7 @@ row_merge_buf_encode(
 	ulint	size;
 	ulint	extra_size;
 
-	size = rec_get_converted_size_temp(
+	size = rec_get_converted_size_temp<false>(
 		index, entry->fields, n_fields, &extra_size);
 	ut_ad(size >= extra_size);
 
@@ -321,7 +321,7 @@ row_merge_buf_encode(
 		*(*b)++ = (byte) (extra_size + 1);
 	}
 
-	rec_convert_dtuple_to_temp(*b + extra_size, index,
+	rec_convert_dtuple_to_temp<false>(*b + extra_size, index,
 				   entry->fields, n_fields);
 
 	*b += size;
@@ -796,7 +796,7 @@ row_merge_buf_add(
 		ulint	size;
 		ulint	extra;
 
-		size = rec_get_converted_size_temp(
+		size = rec_get_converted_size_temp<false>(
 			index, entry->fields, n_fields, &extra);
 
 		ut_ad(data_size + extra_size == size);