Cleanup: Revert unnecessary changes and simplify some code

Revert any changes that are not needed any more.

Most of the code submitted by Vinchen has been refactored so far.
The redo log format changes are as they were submitted by Vinchen.

FIXME: On rollback, convert the record to "old" format if the values
match the 'default row' ones. There should be no need to write any
new BLOBs during rollback. If we created new BLOBs during rollback,
crash recovery could become very tricky.

TODO: Review the changes to row0upd.cc, row0umod.cc. For new inserts
and updates, if the suffix of instantly added columns match the
'default row', we can omit those fields from the clustered index
record. If a non-default value is assigned to the last column, then
all columns will have to be materialized.

TODO: Implement update-in-place of the 'default row' record
(for performing instant ADD COLUMN after instant ADD COLUMN).

TODO: Implement and test rollback of the 'default row' record.

TODO: Test crash recovery. Ensure that the ADD COLUMN operation
is being rolled back if the server is killed during it. Also test
crash recovery of DML operations on tables that were subject to
instant ADD COLUMN, using all supported ROW_FORMAT (REDUNDANT,
COMPACT, DYNAMIC).

storage/innobase/dict/dict0mem.cc

@@ -133,9 +133,9 @@ dict_mem_table_create(
 	table->n_cols = table->n_t_cols - table->n_v_cols;
 
 	table->cols = static_cast<dict_col_t*>(
-		mem_heap_zalloc(heap, table->n_cols * sizeof(dict_col_t)));
+		mem_heap_alloc(heap, table->n_cols * sizeof(dict_col_t)));
 	table->v_cols = static_cast<dict_v_col_t*>(
-		mem_heap_zalloc(heap, n_v_cols * sizeof(*table->v_cols)));
+		mem_heap_alloc(heap, n_v_cols * sizeof(*table->v_cols)));
 
 	/* true means that the stats latch will be enabled -
 	dict_table_stats_lock() will not be noop. */

storage/innobase/handler/handler0alter.cc

@@ -60,8 +60,6 @@ Smart ALTER TABLE
 //#include "wsrep_api.h"
 #include <sql_acl.h>	// PROCESS_ACL
 #endif
-#include "sql_time.h"
-#include "sql_class.h"
 
 static const char *MSG_UNSUPPORTED_ALTER_ONLINE_ON_VIRTUAL_COLUMN=
 			"INPLACE ADD or DROP of virtual columns cannot be "
@@ -4139,13 +4137,14 @@ innobase_add_virtual_try(
 	return(false);
 }
 
-/** Update system table for instant adding column(s)
+/** Insert into SYS_COLUMNS and insert/update the 'default row'
+for instant ADD COLUMN.
 @param[in,out]	ha_alter_info	Data used during in-place alter
 @param[in]	altered_table	MySQL table that is being altered
 @param[in]	table		MySQL table as it is before the ALTER operation
 @param[in,out]	trx		dictionary transaction
-@retval true Failure
-@retval false Success */
+@retval	true	failure
+@retval	false	success */
 static
 bool
 innobase_add_instant_try(

storage/innobase/handler/i_s.cc

@@ -277,30 +277,6 @@ field_store_string(
 	return(ret);
 }
 
-/*******************************************************************//**
-Auxiliary function to store char* value in MYSQL_TYPE_BLOB field.
-@return 0 on success */
-int
-field_store_blob(
-/*===============*/
-	Field*		field,	/*!< in/out: target field for storage */
-	const char*	str,	/*!< in: blob string, or NULL */
-	ulint		str_len)/*!< in: length of blob string */
-{
-	int	ret;
-
-	if (str != NULL) {
-		ret = field->store(str, str_len, 
-				field->charset());
-		field->set_notnull();
-	} else {
-		ret = 0; /* success */
-		field->set_null();
-	}
-
-	return(ret);
-}
-
 /*******************************************************************//**
 Store the name of an index in a MYSQL_TYPE_VARCHAR field.
 Handles the names of incomplete secondary indexes.

storage/innobase/handler/i_s.h

@@ -153,14 +153,4 @@ field_store_string(
 	const char*	str);	/*!< in: NUL-terminated utf-8 string,
 				or NULL */
 
-/*******************************************************************//**
-Auxiliary function to store char* value in MYSQL_TYPE_BLOB field.
-@return 0 on success */
-int
-field_store_blob(
-/*===============*/
-	Field*		field,	/*!< in/out: target field for storage */
-	const char*	str,	/*!< in: blob string, or NULL */
-	ulint		str_len);/*!< in: length of blob string */
-
 #endif /* i_s_h */

storage/innobase/include/dict0dict.ic

@@ -536,10 +536,8 @@ dict_table_get_nth_v_col(
 	ut_ad(table);
 	ut_ad(pos < table->n_v_def);
 	ut_ad(table->magic_n == DICT_TABLE_MAGIC_N);
-	dict_v_col_t* vcol = &table->v_cols[pos];
-	ut_ad(vcol->m_col.def_val.len == UNIV_SQL_DEFAULT);
-	ut_ad(!vcol->m_col.def_val.data);
-	return vcol;
+	ut_ad(!table->v_cols[pos].m_col.is_instant());
+	return &table->v_cols[pos];
 }
 
 /********************************************************************//**

storage/innobase/include/rem0rec.ic

@@ -926,9 +926,9 @@ rec_get_nth_field_offs(
 
 	if (length & REC_OFFS_SQL_NULL) {
 		length = UNIV_SQL_NULL;
-	} else if ( length & REC_OFFS_DEFAULT) { 
+	} else if (length & REC_OFFS_DEFAULT) {
 		length = UNIV_SQL_DEFAULT;
-	}else {
+	} else {
 		length &= REC_OFFS_MASK;
 		length -= offs;
 	}

storage/innobase/mtr/mtr0log.cc

@@ -436,7 +436,7 @@ mlog_open_and_write_index(
 		log_end = log_ptr + 11 + size;
 	} else {
 		ulint	i;
-		ibool is_instant = index->is_instant();
+		bool	is_instant = index->is_instant();
 		ulint	n	= dict_index_get_n_fields(index);
 		ulint	total	= 11 + (is_instant ? 2 : 0) + size + (n + 2) * 2;
 		ulint	alloc	= total;
@@ -472,7 +472,6 @@ mlog_open_and_write_index(
 
 			// record the n_core_fields
 			mach_write_to_2(log_ptr, index->n_core_fields);
-
 		} else {
 			mach_write_to_2(log_ptr, n);
 		}
@@ -563,13 +562,17 @@ mlog_parse_index(
 		}
 		n = mach_read_from_2(ptr);
 		ptr += 2;
-		if (n & 0x8000) {  // instant record
+		if (n & 0x8000) { /* record after instant ADD COLUMN */
 			n &= 0x7FFF;
 
 			n_core_fields = mach_read_from_2(ptr);
+
+			if (!n_core_fields || n_core_fields > n) {
+				recv_sys->found_corrupt_log = TRUE;
+				return(NULL);
+			}
+
 			ptr += 2;
-			ut_ad(n_core_fields > 0);
-			ut_ad(n_core_fields <= n);
 
 			if (end_ptr < ptr + 2) {
 				return(NULL);

storage/innobase/page/page0cur.cc

@@ -259,7 +259,6 @@ page_cur_rec_field_extends(
 	const dtuple_t*	tuple,	/*!< in: data tuple */
 	const rec_t*	rec,	/*!< in: record */
 	const ulint*	offsets,/*!< in: array returned by rec_get_offsets() */
-	const dict_index_t*		index, /*!< in: index of rec */
 	ulint		n)	/*!< in: compare nth field */
 {
 	const dtype_t*	type;
@@ -443,7 +442,7 @@ page_cur_search_with_match(
 #ifdef PAGE_CUR_LE_OR_EXTENDS
 			if (mode == PAGE_CUR_LE_OR_EXTENDS
 			    && page_cur_rec_field_extends(
-				    tuple, mid_rec, offsets, index,
+				    tuple, mid_rec, offsets,
 				    cur_matched_fields)) {
 
 				goto low_slot_match;
@@ -497,7 +496,7 @@ page_cur_search_with_match(
 #ifdef PAGE_CUR_LE_OR_EXTENDS
 			if (mode == PAGE_CUR_LE_OR_EXTENDS
 			    && page_cur_rec_field_extends(
-				    tuple, mid_rec, offsets, index, 
+				    tuple, mid_rec, offsets,
 				    cur_matched_fields)) {
 
 				goto low_rec_match;
@@ -692,7 +691,7 @@ page_cur_search_with_match_bytes(
 #ifdef PAGE_CUR_LE_OR_EXTENDS
 			if (mode == PAGE_CUR_LE_OR_EXTENDS
 			    && page_cur_rec_field_extends(
-				    tuple, mid_rec, offsets, index, 
+				    tuple, mid_rec, offsets,
 				    cur_matched_fields)) {
 
 				goto low_slot_match;
@@ -749,7 +748,7 @@ page_cur_search_with_match_bytes(
 #ifdef PAGE_CUR_LE_OR_EXTENDS
 			if (mode == PAGE_CUR_LE_OR_EXTENDS
 			    && page_cur_rec_field_extends(
-				    tuple, mid_rec, offsets, index, 
+				    tuple, mid_rec, offsets,
 				    cur_matched_fields)) {
 
 				goto low_rec_match;

storage/innobase/page/page0zip.cc

@@ -2291,9 +2291,6 @@ page_zip_decompress_heap_no(
 	/* Set heap_no and the status bits. */
 	mach_write_to_2(rec - REC_NEW_HEAP_NO, heap_status);
 	heap_status += 1 << REC_HEAP_NO_SHIFT;
-
-	/* Set the info bit to zero for instant add columns */
-	rec[-REC_N_NEW_EXTRA_BYTES] = 0;
 	return(TRUE);
 }
 

storage/innobase/rem/rem0cmp.cc

@@ -409,7 +409,8 @@ cmp_data(
 	const byte*	data2,
 	ulint		len2)
 {
-	ut_ad(len1 != UNIV_SQL_DEFAULT && len2 != UNIV_SQL_DEFAULT);
+	ut_ad(len1 != UNIV_SQL_DEFAULT);
+	ut_ad(len2 != UNIV_SQL_DEFAULT);
 
 	if (len1 == UNIV_SQL_NULL || len2 == UNIV_SQL_NULL) {
 		if (len1 == len2) {
@@ -710,8 +711,10 @@ cmp_dtuple_rec_with_match_low(
 		contain externally stored fields, and the first fields
 		(primary key fields) should already differ. */
 		ut_ad(!rec_offs_nth_extern(offsets, cur_field));
-		
-		/* We should never compare against instant add columns */
+		/* We should never compare against instantly added columns.
+		Columns can only be instantly added to clustered index
+		leaf page records, and the first fields (primary key fields)
+		should already differ. */
 		ut_ad(!rec_offs_nth_default(offsets, cur_field));
 
 		rec_b_ptr = rec_get_nth_field(rec, offsets, cur_field,

storage/innobase/rem/rem0rec.cc

@@ -1465,21 +1465,23 @@ rec_convert_dtuple_to_rec_old(
 	return(rec);
 }
 
-/*********************************************************//**
-Builds a ROW_FORMAT=COMPACT record out of a data tuple. 
-@return true if instant record */
-UNIV_INLINE
-bool
+/** Convert a data tuple into a ROW_FORMAT=COMPACT record.
+@param[out]	rec		converted record
+@param[in]	index		index
+@param[in]	fields		data fields to convert
+@param[in]	n_fields	number of data fields
+@param[in]	status		rec_get_status(rec)
+@param[in]	temp		whether to use the format for temporary files
+				in index creation */
+static inline
+void
 rec_convert_dtuple_to_rec_comp(
-/*===========================*/
-	rec_t*			rec,	/*!< in: origin of record */
-	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 */
-	rec_comp_status_t	status,	/*!< in: status bits of the record */
-	bool			temp)	/*!< in: whether to use the
-					format for temporary files in
-					index creation */
+	rec_t*			rec,
+	const dict_index_t*	index,
+	const dfield_t*		fields,
+	ulint			n_fields,
+	rec_comp_status_t	status,
+	bool			temp)
 {
 	const dfield_t*	field;
 	const dtype_t*	type;
@@ -1491,7 +1493,6 @@ rec_convert_dtuple_to_rec_comp(
 	ulint		n_node_ptr_field;
 	ulint		fixed_len;
 	ulint		null_mask	= 1;
-	bool		is_instant = false;
 
 	ut_ad(n_fields > 0);
 	ut_ad(temp || dict_table_is_comp(index->table));
@@ -1514,25 +1515,27 @@ rec_convert_dtuple_to_rec_comp(
 		rec_set_heap_no_new(rec, PAGE_HEAP_NO_USER_LOW);
 		nulls = rec - (REC_N_NEW_EXTRA_BYTES + 1);
 
-		switch (UNIV_EXPECT(status, REC_STATUS_ORDINARY)) {
+		switch (status) {
 		case REC_STATUS_COLUMNS_ADDED:
 			ut_ad(index->is_instant());
 			/* fall through */
 		case REC_STATUS_ORDINARY:
 			ut_ad(n_fields <= dict_index_get_n_fields(index));
-			is_instant = index->is_instant();
-			if (is_instant) {
+			if (index->is_instant()) {
 				ut_ad(n_fields == index->n_fields);
 				rec_set_status(rec, REC_STATUS_COLUMNS_ADDED);
 				nulls = rec_set_n_add_field(
 					rec, n_fields - 1
 					- index->n_core_fields);
+			} else {
+				rec_set_status(rec, REC_STATUS_ORDINARY);
 			}
 
 			n_node_ptr_field = ULINT_UNDEFINED;
 			lens = nulls - UT_BITS_IN_BYTES(index->n_nullable);
 			break;
 		case REC_STATUS_NODE_PTR:
+			rec_set_status(rec, status);
 			ut_ad(n_fields
 			      == dict_index_get_n_unique_in_tree_nonleaf(index)
 				 + 1);
@@ -1543,14 +1546,8 @@ rec_convert_dtuple_to_rec_comp(
 			break;
 		case REC_STATUS_INFIMUM:
 		case REC_STATUS_SUPREMUM:
-			ut_ad(n_fields == 1);
-			n_node_ptr_field = ULINT_UNDEFINED;
-			lens = nulls;
-			ut_d(n_null = 0);
-			break;
-		default:
 			ut_error;
-			return false;
+			return;
 		}
 	}
 
@@ -1651,8 +1648,6 @@ rec_convert_dtuple_to_rec_comp(
 			end += len;
 		}
 	}
-
-	return is_instant;
 }
 
 /** Write indexed virtual column data into a temporary file.
@@ -1730,30 +1725,20 @@ rec_convert_dtuple_to_rec_new(
 	const dict_index_t*	index,	/*!< in: record descriptor */
 	const dtuple_t*		dtuple)	/*!< in: data tuple */
 {
-	ulint bits = dtuple->info_bits;
-	ut_ad(!(bits & ~(REC_NEW_STATUS_MASK | REC_INFO_DELETED_FLAG
-			 | REC_INFO_MIN_REC_FLAG)));
+	ut_ad(!(dtuple->info_bits
+		& ~(REC_NEW_STATUS_MASK | REC_INFO_DELETED_FLAG
+		    | REC_INFO_MIN_REC_FLAG)));
 	rec_comp_status_t status = static_cast<rec_comp_status_t>(
-		bits & REC_NEW_STATUS_MASK);
+		dtuple->info_bits & REC_NEW_STATUS_MASK);
 	ulint	extra_size;
 
 	rec_get_converted_size_comp(
 		index, status, dtuple->fields, dtuple->n_fields, &extra_size);
 	rec_t* rec = buf + extra_size;
 
-	if (rec_convert_dtuple_to_rec_comp(
-		    rec, index, dtuple->fields, dtuple->n_fields,
-		    status, false)) {
-		ut_ad(status == REC_STATUS_ORDINARY
-		      || status == REC_STATUS_COLUMNS_ADDED);
-		bits &= ~REC_NEW_STATUS_MASK;
-		bits |= REC_STATUS_COLUMNS_ADDED;
-	} else {
-		ut_ad(status == REC_STATUS_ORDINARY
-		      || status == REC_STATUS_NODE_PTR);
-	}
-
-	rec_set_info_and_status_bits(rec, bits);
+	rec_convert_dtuple_to_rec_comp(
+		rec, index, dtuple->fields, dtuple->n_fields, status, false);
+	rec_set_info_bits_new(rec, dtuple->info_bits & ~REC_NEW_STATUS_MASK);
 	return(rec);
 }
 
@@ -1862,10 +1847,10 @@ rec_copy_prefix_to_dtuple(
 				  n_fields, &heap);
 
 	ut_ad(rec_validate(rec, offsets));
+	ut_ad(!rec_offs_any_default(offsets));
 	ut_ad(dtuple_check_typed(tuple));
 
-	dtuple_set_info_bits(tuple, rec_get_info_bits(
-				     rec, dict_table_is_comp(index->table)));
+	tuple->info_bits = rec_get_info_bits(rec, rec_offs_comp(offsets));
 
 	for (ulint i = 0; i < n_fields; i++) {
 		dfield_t*	field;
@@ -1883,8 +1868,6 @@ rec_copy_prefix_to_dtuple(
 			dfield_set_null(field);
 		}
 	}
-
-	ut_ad(!rec_offs_any_default(offsets));
 }
 
 /**************************************************************//**
@@ -1950,7 +1933,6 @@ rec_copy_prefix_to_buf(
 	ulint		i;
 	ulint		prefix_len;
 	ulint		null_mask;
-	ulint		status;
 	bool		is_rtr_node_ptr = false;
 
 	ut_ad(index->n_core_null_bytes <= UT_BITS_IN_BYTES(index->n_nullable));
@@ -1964,9 +1946,7 @@ rec_copy_prefix_to_buf(
 			       buf, buf_size));
 	}
 
-	status = rec_get_status(rec);
-
-	switch (status) {
+	switch (rec_get_status(rec)) {
 	case REC_STATUS_COLUMNS_ADDED:
 		ut_ad(index->is_instant());
 		if (n_fields >= index->n_core_fields) {
@@ -2150,13 +2130,13 @@ rec_validate(
 	for (i = 0; i < n_fields; i++) {
 		rec_get_nth_field_offs(offsets, i, &len);
 
-		if (!((len < UNIV_PAGE_SIZE) || (len == UNIV_SQL_NULL) || (len == UNIV_SQL_DEFAULT))) {
-			ib::error() << "Record field " << i << " len " << len;
-			return(FALSE);
-		}
-
 		switch (len) {
 		default:
+			if (len >= UNIV_PAGE_SIZE) {
+				ib::error() << "Record field " << i
+					<< " len " << len;
+				return(FALSE);
+			}
 			len_sum += len;
 			break;
 		case UNIV_SQL_DEFAULT:
@@ -2279,7 +2259,7 @@ rec_print_comp(
 				fprintf(file, " (total " ULINTPF " bytes)",
 					len);
 			}
-		} 
+		}
 		putc(';', file);
 		putc('\n', file);
 	}
@@ -2372,6 +2352,7 @@ rec_print_mbr_rec(
 	ut_ad(rec);
 	ut_ad(offsets);
 	ut_ad(rec_offs_validate(rec, NULL, offsets));
+	ut_ad(!rec_offs_any_default(offsets));
 
 	if (!rec_offs_comp(offsets)) {
 		rec_print_mbr_old(file, rec);
@@ -2382,11 +2363,7 @@ rec_print_mbr_rec(
 		const byte*	data;
 		ulint		len;
 
-		if (rec_offs_nth_default(offsets, i)) {
-			len = UNIV_SQL_DEFAULT;
-		} else {
-			data = rec_get_nth_field(rec, offsets, i, &len);
-		}
+		data = rec_get_nth_field(rec, offsets, i, &len);
 
 		if (i == 0) {
 			fprintf(file, " MBR:");
@@ -2404,11 +2381,7 @@ rec_print_mbr_rec(
 		} else {
 			fprintf(file, " %lu:", (ulong) i);
 
-			if (len == UNIV_SQL_NULL) {
-				fputs(" SQL NULL", file);
-			} else if (len == UNIV_SQL_DEFAULT) {
-				fputs(" SQL DEFAULT", file);
-			} else {
+			if (len != UNIV_SQL_NULL) {
 				if (len <= 30) {
 
 					ut_print_buf(file, data, len);
@@ -2418,6 +2391,8 @@ rec_print_mbr_rec(
 					fprintf(file, " (total %lu bytes)",
 						(ulong) len);
 				}
+			} else {
+				fputs(" SQL NULL", file);
 			}
 		}
 		putc(';', file);
@@ -2527,17 +2502,17 @@ rec_print(
 			o << ',';
 		}
 
-		if (rec_offs_nth_default(offsets, i)) {
+		data = rec_get_nth_field(rec, offsets, i, &len);
+
+		if (len == UNIV_SQL_DEFAULT) {
 			o << "DEFAULT";
 			continue;
 		}
 
-		data = rec_get_nth_field(rec, offsets, i, &len);
-
 		if (len == UNIV_SQL_NULL) {
 			o << "NULL";
 			continue;
-		} 
+		}
 
 		if (rec_offs_nth_extern(offsets, i)) {
 			ulint	local_len = len - BTR_EXTERN_FIELD_REF_SIZE;