MDEV-15662 Instant DROP COLUMN or changing the order of columns

Instant drop column works for dynamic row format as of now.

- Yet to change default row for redundant row format.
- Need to find better condition for is_instant()
- Need to find better way to store offsets for new default row.
- Yet to do rollback for instant alter drop operation.

mysql-test/suite/innodb/t/instant_drop.test

+--source include/have_innodb.inc
+create table t1(f1 int not null, f2 int not null, f3 int not null)engine=innodb;
+insert into t1 values(1, 2, 3),(4, 5, 6);
+alter table t1 drop column f2, algorithm=instant;
+select * from t1;
+insert into t1 values(1,2);
+select * from t1;
+alter table t1 add column f4 int not null default 5, algorithm=instant;
+select * from t1;
+alter table t1 drop column f1, algorithm=instant;
+select * from t1;
+insert into t1 values(7, 9);
+select * from t1;
+alter table t1 add column f5 blob default repeat('aaa', 950), drop column f4, algorithm=instant;
+select * from t1;
+select f3 from t1;
+update t1 set f3 = 10 where f3 > 2;
+select * from t1;
+delete from t1 where f3 = 10;
+show create table t1;
+select f3 from t1;
+update t1 set f5 = "world";
+select * from t1;
+drop table t1;
+
+create table t1(f1 int, f2 int not null, index idx(f2))engine=innodb;
+insert into t1 values(1, 2);
+alter table t1 drop column f1, add column f3 varchar(100) default "thiru", algorithm=instant;
+select * from t1 force index (idx);
+alter table t1 drop column f3, algorithm=instant;
+select * from t1;
+begin;
+insert into t1 values(10);
+select * from t1;
+update t1 set f2 = 100;
+select * from t1;
+delete from t1 where f2 = 100;
+select * from t1;
+rollback;
+select * from t1;
+show create table t1;
+drop table t1;
+
+create table t1(f1 int, f2 int not null)engine=innodb;
+insert into t1 values(1, 2);
+alter table t1 drop column f2, algorithm=instant;
+insert into t1 values(NULL);
+select * from t1;
+drop table t1;
+
+create table t1(f1 int not null, f2 int not null)engine=innodb;
+insert into t1 values(1, 2);
+alter table t1 add column f5 int default 10, algorithm=instant;
+alter table t1 add column f3 int not null default 100, algorithm=instant;
+alter table t1 add column f4 int default 100, drop column f3, algorithm=instant;
+insert into t1 values(2, 3, 20, 100);
+select * from t1;
+drop table t1;
+
+create table t1(f1 int not null, f2 int not null) engine=innodb;
+insert into t1 values(1, 1);
+alter table t1 drop column f2, add column f3 int default 3, algorithm=instant;
+select * from t1;
+update t1 set f3 = 19;
+select * from t1;
+alter table t1 drop column f1, add column f5 int default 10, algorithm=instant;
+insert into t1 values(4, 10);
+select * from t1;
+--source include/restart_mysqld.inc
+select * from t1;
+alter table t1 add column f6 int default 9,drop column f5, algorithm = instant;
+insert into t1 values(4, 9);
+alter table t1 force, algorithm=inplace;
+select * from t1;
+drop table t1;
+
+create table t1(f1 int, f2 int not null) engine=innodb;
+insert into t1(f1, f2) values(1, 2);
+select * from information_schema.innodb_sys_columns;
+alter table t1 drop column f2, add column f4 varchar(100) default repeat('a', 20), add column f5 int default 10,  algorithm=instant;
+select * from t1;
+show create table t1;
+alter table t1 add column f6 char(100) default repeat('a', 99), algorithm=instant;
+--source include/restart_mysqld.inc
+select * from t1;
+alter table t1 force, algorithm=inplace;
+select * from t1;
+show create table t1;
+drop table t1;
+
+create table t1(f1 int, f2 int not null)engine=innodb;
+insert into t1 values(1, 2);
+alter table t1 drop column f2, add column f3 int default 1, add column f4 int default 4, algorithm=instant;
+--source include/restart_mysqld.inc
+select * from t1;
+alter table t1 add column f5 char(100) default repeat('a', 99), algorithm=instant;
+--source include/restart_mysqld.inc
+select * from t1;
+drop table t1;

storage/innobase/data/data0data.cc

@@ -43,8 +43,8 @@ byte	data_error;
 #endif /* UNIV_DEBUG */
 
 /** Trim the tail of an index tuple before insert or update.
-After instant ADD COLUMN, if the last fields of a clustered index tuple
-match the 'default row', there will be no need to store them.
+After instant COLUMN operation, if the last fields of a clustered index tuple
+match the 'default row' or it is dropped, there will be no need to store them.
 NOTE: A page latch in the index must be held, so that the index
 may not lose 'instantness' before the trimmed tuple has been
 inserted or updated.
@@ -59,7 +59,12 @@ void dtuple_t::trim(const dict_index_t& index)
 	for (; i > index.n_core_fields; i--) {
 		const dfield_t* dfield = dtuple_get_nth_field(this, i - 1);
 		const dict_col_t* col = dict_index_get_nth_col(&index, i - 1);
-		ut_ad(col->is_instant());
+
+		if (col->is_dropped()) {
+			continue;
+		}
+
+		ut_ad(col->is_instant_add());
 		ulint len = dfield_get_len(dfield);
 		if (len != col->def_val.len) {
 			break;
@@ -75,6 +80,13 @@ void dtuple_t::trim(const dict_index_t& index)
 	n_fields = i;
 }
 
+/** Whether the dtuple represnets default row with drop column info.
+@return true if it is default row with drop column or false. */
+bool dtuple_t::is_new_default_row() const
+{
+	return info_bits == REC_INFO_DEFAULT_ROW_DROP;
+}
+
 /** Compare two data tuples.
 @param[in] tuple1 first data tuple
 @param[in] tuple2 second data tuple
@@ -638,23 +650,38 @@ dtuple_convert_big_rec(
 
 	n_fields = 0;
 
+	bool	drop_column_blob = true;
+	bool	new_default_row = entry->is_new_default_row();
+
 	while (page_zip_rec_needs_ext(rec_get_converted_size(index, entry,
 							     *n_ext),
 				      dict_table_is_comp(index->table),
 				      dict_index_get_n_fields(index),
-				      dict_table_page_size(index->table))) {
-
+				      dict_table_page_size(index->table))
+	       || UNIV_UNLIKELY(new_default_row && drop_column_blob)) {
 		ulint			i;
 		ulint			longest		= 0;
 		ulint			longest_i	= ULINT_MAX;
 		byte*			data;
+		ulint			field_no	= 0;
 
 		for (i = dict_index_get_n_unique_in_tree(index);
 		     i < dtuple_get_n_fields(entry); i++) {
 			ulint	savings;
 
 			dfield = dtuple_get_nth_field(entry, i);
-			ifield = dict_index_get_nth_field(index, i);
+
+			if (UNIV_UNLIKELY(new_default_row
+				&& i == unsigned(index->n_uniq + DATA_ROLL_PTR))) {
+				longest_i = i;
+				dfield->data =
+					index->table->construct_default_row_blob(
+						heap, &dfield->len);
+				drop_column_blob = false;
+				goto ext_write;
+			}
+
+			ifield = dict_index_get_nth_field(index, field_no++);
 
 			/* Skip fixed-length, NULL, externally stored,
 			or short columns */
@@ -709,9 +736,8 @@ dtuple_convert_big_rec(
 		We store the first bytes locally to the record. Then
 		we can calculate all ordering fields in all indexes
 		from locally stored data. */
-
+ext_write:
 		dfield = dtuple_get_nth_field(entry, longest_i);
-		ifield = dict_index_get_nth_field(index, longest_i);
 		local_prefix_len = local_len - BTR_EXTERN_FIELD_REF_SIZE;
 
 		vector->append(
@@ -736,7 +762,6 @@ dtuple_convert_big_rec(
 		UNIV_MEM_ALLOC(data + local_prefix_len,
 			       BTR_EXTERN_FIELD_REF_SIZE);
 #endif
-
 		dfield_set_data(dfield, data, local_len);
 		dfield_set_ext(dfield);
 

storage/innobase/dict/dict0dict.cc

@@ -2788,6 +2788,7 @@ dict_index_add_col(
 
 	if (!(col->prtype & DATA_NOT_NULL)) {
 		index->n_nullable++;
+		index->n_non_drop_nullable_fields++;
 	}
 }
 

storage/innobase/include/data0data.h

@@ -607,6 +607,15 @@ struct dfield_t{
 		ut_ad(0);
 		return false;
 	}
+
+	void init_new_default_blob(void* value, ulint val_len)
+	{
+		data = value;
+		len = val_len;
+		type.mtype = DATA_BLOB;
+		type.prtype = DATA_NOT_NULL;
+		ext = 1;
+	}
 };
 
 /** Structure for an SQL data tuple of fields (logical record) */
@@ -644,6 +653,9 @@ struct dtuple_t {
 	inserted or updated.
 	@param[in]	index	index possibly with instantly added columns */
 	void trim(const dict_index_t& index);
+
+	/** Default row data tuple with drop column information.*/
+	bool is_new_default_row() const;
 };
 
 /** A slot for a field in a big rec vector */

storage/innobase/include/dict0dict.h

@@ -1112,6 +1112,20 @@ dict_index_get_n_fields(
 					the dictionary cache) */
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 
+/** Get the number of the dropped fields in the internal representation of
+an index.
+@param[in]	index	internal representation of index
+@return number of dropped fields. */
+UNIV_INLINE
+ulint dict_index_get_n_dropped_fields(const dict_index_t* index);
+
+/** Get the number of non-dropped fields in the internal representation of
+an index.
+@param[in]	index	internal representation of index
+@return number of non-dropped fields. */
+UNIV_INLINE
+ulint dict_index_get_n_non_dropped_fields(const dict_index_t* index);
+
 /********************************************************************//**
 Gets the number of fields in the internal representation of an index
 that uniquely determine the position of an index entry in the index, if

storage/innobase/include/dict0dict.ic

@@ -490,7 +490,7 @@ 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);
-	ut_ad(!table->v_cols[pos].m_col.is_instant());
+	ut_ad(!table->v_cols[pos].m_col.is_instant_add());
 	return &table->v_cols[pos];
 }
 
@@ -916,6 +916,26 @@ dict_index_get_n_fields(
 	return(index->n_fields);
 }
 
+/** Get the number of the dropped fields in the internal representation of
+an index.
+@param[in]	index	internal representation of index
+@return number of dropped fields. */
+UNIV_INLINE
+ulint dict_index_get_n_dropped_fields(const dict_index_t* index)
+{
+	ut_ad(index);
+	ut_ad(index->magic_n == DICT_INDEX_MAGIC_N);
+	return index->n_dropped_fields;
+}
+
+UNIV_INLINE
+ulint dict_index_get_n_non_dropped_fields(const dict_index_t* index)
+{
+	ut_ad(index);
+	ut_ad(index->magic_n == DICT_INDEX_MAGIC_N);
+	return index->n_fields - index->n_dropped_fields;
+}
+
 /********************************************************************//**
 Gets the number of fields in the internal representation of an index
 that uniquely determine the position of an index entry in the index, if

storage/innobase/include/dict0mem.ic

@@ -64,6 +64,9 @@ dict_mem_fill_index_struct(
 	/* The '1 +' above prevents allocation
 	of an empty mem block */
 	index->nulls_equal = false;
+	index->n_dropped_fields = 0;
+	index->n_non_drop_nullable_fields = 0;
+	index->instant = false;
 #ifdef BTR_CUR_HASH_ADAPT
 #ifdef MYSQL_INDEX_DISABLE_AHI
 	index->disable_ahi = false;

storage/innobase/include/rem0rec.h

@@ -67,13 +67,19 @@ enum rec_comp_status_t {
 	REC_STATUS_SUPREMUM = 3,
 	/** Clustered index record that has been inserted or updated
 	after instant ADD COLUMN (more than dict_index_t::n_core_fields) */
-	REC_STATUS_COLUMNS_ADDED = 4
+	REC_STATUS_COLUMNS_INSTANT = 4
 };
 
 /** The dtuple_t::info_bits of the 'default row' record.
 @see rec_is_default_row() */
 static const byte REC_INFO_DEFAULT_ROW
-	= REC_INFO_MIN_REC_FLAG | REC_STATUS_COLUMNS_ADDED;
+	= REC_INFO_MIN_REC_FLAG | REC_STATUS_COLUMNS_INSTANT;
+
+/** The dtuple_t::info_bits of the 'default row' record with dropped
+column information. */
+static const byte REC_INFO_DEFAULT_ROW_DROP
+	= REC_INFO_MIN_REC_FLAG | REC_INFO_DELETED_FLAG
+	  | REC_STATUS_COLUMNS_INSTANT;
 
 #define REC_NEW_STATUS		3	/* This is single byte bit-field */
 #define REC_NEW_STATUS_MASK	0x7UL
@@ -124,10 +130,16 @@ const ulint REC_OFFS_COMPACT = ~(ulint(~0) >> 1);
 const ulint REC_OFFS_SQL_NULL = REC_OFFS_COMPACT;
 /** External flag in offsets returned by rec_get_offsets() */
 const ulint REC_OFFS_EXTERNAL = REC_OFFS_COMPACT >> 1;
+
 /** Default value flag in offsets returned by rec_get_offsets() */
 const ulint REC_OFFS_DEFAULT = REC_OFFS_COMPACT >> 2;
+
+const ulint REC_OFFS_DROP_COL = REC_OFFS_COMPACT >> 3;
+
+const ulint REC_OFFS_DROP_SQL_NULL = REC_OFFS_COMPACT >> 4;
+
 /** Mask for offsets returned by rec_get_offsets() */
-const ulint REC_OFFS_MASK = REC_OFFS_DEFAULT - 1;
+const ulint REC_OFFS_MASK = REC_OFFS_DROP_SQL_NULL - 1;
 
 #ifndef UNIV_INNOCHECKSUM
 /******************************************************//**
@@ -296,7 +308,7 @@ rec_comp_status_t
 rec_get_status(const rec_t* rec)
 {
 	byte bits = rec[-REC_NEW_STATUS] & REC_NEW_STATUS_MASK;
-	ut_ad(bits <= REC_STATUS_COLUMNS_ADDED);
+	ut_ad(bits <= REC_STATUS_COLUMNS_INSTANT);
 	return static_cast<rec_comp_status_t>(bits);
 }
 
@@ -307,7 +319,7 @@ inline
 void
 rec_set_status(rec_t* rec, byte bits)
 {
-	ut_ad(bits <= REC_STATUS_COLUMNS_ADDED);
+	ut_ad(bits <= REC_STATUS_COLUMNS_INSTANT);
 	rec[-REC_NEW_STATUS] = (rec[-REC_NEW_STATUS] & ~REC_NEW_STATUS_MASK)
 		| bits;
 }
@@ -711,6 +723,20 @@ rec_offs_nth_sql_null(const ulint* offsets, ulint n)
 	return rec_offs_nth_flag(offsets, n, REC_OFFS_SQL_NULL);
 }
 
+inline
+ulint
+rec_offs_nth_drop_sql_null(const ulint* offsets, ulint n)
+{
+	return rec_offs_nth_flag(offsets, n, REC_OFFS_DROP_SQL_NULL);
+}
+
+inline
+ulint
+rec_offs_nth_drop_col(const ulint* offsets, ulint n)
+{
+	return rec_offs_nth_flag(offsets, n, REC_OFFS_DROP_COL);
+}
+
 /** Determine if a record field is stored off-page.
 @param[in]	offsets	rec_get_offsets()
 @param[in]	n	nth field
@@ -779,10 +805,52 @@ rec_is_default_row(const rec_t* rec, const dict_index_t* index)
 		& REC_INFO_MIN_REC_FLAG;
 	ut_ad(!is || index->is_instant());
 	ut_ad(!is || !dict_table_is_comp(index->table)
-	      || rec_get_status(rec) == REC_STATUS_COLUMNS_ADDED);
+	      || rec_get_status(rec) == REC_STATUS_COLUMNS_INSTANT);
+	return is;
+}
+
+/** Determine if the record is the 'new default row' pseudo-record
+in the clustered index.
+@param[in]	rec	leaf page record
+@param[in]	index	index of the record
+@return	whether the record is the 'new default row' pseudo-record */
+inline
+bool
+rec_is_new_default_row(const rec_t* rec, const dict_index_t* index)
+{
+	bool is = rec_get_info_bits(rec, dict_table_is_comp(index->table))
+			& REC_INFO_MIN_REC_FLAG;
+	is = is && rec_get_deleted_flag(rec, dict_table_is_comp(index->table));
+	ut_ad(!is || index->is_instant());
+	ut_ad(!is || !dict_table_is_comp(index->table)
+	      || rec_get_status(rec) == REC_STATUS_COLUMNS_INSTANT);
 	return is;
 }
 
+/** Get the nth field from an index.
+@param[in]	rec	index record
+@param[in]	index	index
+@param[in]	offsets	rec_get_offsets(rec, index)
+@param[in]	n	field number
+@param[out]	len	length of the field in bytes, or UNIV_SQL_NULL
+@return a read-only copy of the index field */
+inline
+const byte*
+rec_get_nth_def_field(
+	const rec_t*		rec,
+	const dict_index_t*	index,
+	const ulint*		offsets,
+	ulint			n,
+	ulint*			len)
+{
+	ut_ad(rec_offs_validate(rec, index, offsets));
+	if (!rec_offs_nth_default(offsets, n)) {
+		return rec_get_nth_field(rec, offsets, n, len);
+	}
+
+	return index->instant_field_value(n - 1, len);
+}
+
 /** Get the nth field from an index.
 @param[in]	rec	index record
 @param[in]	index	index
@@ -800,6 +868,12 @@ rec_get_nth_cfield(
 	ulint*			len)
 {
 	ut_ad(rec_offs_validate(rec, index, offsets));
+
+	if (rec_offs_nth_drop_col(offsets, n)) {
+		*len = index->fields[n].col->len;
+		return field_ref_zero;
+	}
+
 	if (!rec_offs_nth_default(offsets, n)) {
 		return rec_get_nth_field(rec, offsets, n, len);
 	}
@@ -946,7 +1020,7 @@ rec_copy(
 @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_COLUMNS_ADDED
+@param[in]	status		REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_INSTANT
 @return	total size, in bytes */
 ulint
 rec_get_converted_size_temp(
@@ -962,7 +1036,7 @@ rec_get_converted_size_temp(
 @param[in]	index	index of that the record belongs to
 @param[in,out]	offsets	offsets to the fields; in: rec_offs_n_fields(offsets)
 @param[in]	n_core	number of core fields (index->n_core_fields)
-@param[in]	status	REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_ADDED */
+@param[in]	status	REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_INSTANT */
 void
 rec_init_offsets_temp(
 	const rec_t*		rec,
@@ -988,8 +1062,7 @@ rec_init_offsets_temp(
 @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_COLUMNS_ADDED
-*/
+@param[in]	status		REC_STATUS_ORDINARY or REC_STATUS_COLUMNS_INSTANT */
 void
 rec_convert_dtuple_to_temp(
 	rec_t*			rec,
@@ -1052,21 +1125,44 @@ rec_get_converted_size_comp_prefix(
 	ulint			n_fields,/*!< in: number of data fields */
 	ulint*			extra)	/*!< out: extra size */
 	MY_ATTRIBUTE((warn_unused_result, nonnull(1,2)));
-/**********************************************************//**
-Determines the size of a data tuple in ROW_FORMAT=COMPACT.
+
+/** Determines the size of a data tuple in ROW_FORMAT=COMPACT.
+@param[in]	index		record descriptor. dict_table_is_comp()
+				is assumed to hold, even if it doesn't
+@param[in]	status		status bits of the record
+@param[in]	fields		array of data fields
+@param[in]	n_fields	number of data fields
+@param[out]	extra		extra size
+@param[in]	new_default_row	default row with drop column info
 @return total size */
 ulint
 rec_get_converted_size_comp(
-/*========================*/
-	const dict_index_t*	index,	/*!< in: record descriptor;
-					dict_table_is_comp() is
-					assumed to hold, even if
-					it does not */
-	rec_comp_status_t	status,	/*!< in: status bits of the record */
-	const dfield_t*		fields,	/*!< in: array of data fields */
-	ulint			n_fields,/*!< in: number of data fields */
-	ulint*			extra)	/*!< out: extra size */
+	const dict_index_t*	index,
+	rec_comp_status_t	status,
+	const dfield_t*		fields,
+	ulint			n_fields,
+	ulint*			extra)
 	MY_ATTRIBUTE((nonnull(1,3)));
+
+/** Determines the size of a data tupel in ROW_FORMAT=COMPACT.
+@param[in]	index		clustered index
+@param[in]	fields		array of data fields
+@param[in]	n_fields	number of data fields
+@param[out]	extra		extra size
+@return total size of default row with drop column info. */
+ulint
+rec_get_def_converted_size_comp(
+	const dict_index_t*	index,
+	const dfield_t*		fields,
+	ulint			n_fields,
+	ulint*			extra);
+
+ulint
+rec_get_default_rec_converted_size(
+	const dict_index_t*	clust_index,
+	const dtuple_t*		dtuple,
+	ulint*			ext);
+
 /**********************************************************//**
 The following function returns the size of a data tuple when converted to
 a physical record.

storage/innobase/include/rem0rec.ic

@@ -67,7 +67,7 @@ most significant bytes and bits are written below less significant.
 					001=REC_STATUS_NODE_PTR
 					010=REC_STATUS_INFIMUM
 					011=REC_STATUS_SUPREMUM
-					100=REC_STATUS_COLUMNS_ADDED
+					100=REC_STATUS_COLUMNS_INSTANT
 					1xx=reserved
 				5 bits heap number
 			4	8 bits heap number
@@ -453,7 +453,7 @@ rec_get_n_fields(
 	}
 
 	switch (rec_get_status(rec)) {
-	case REC_STATUS_COLUMNS_ADDED:
+	case REC_STATUS_COLUMNS_INSTANT:
 	case REC_STATUS_ORDINARY:
 		return(dict_index_get_n_fields(index));
 	case REC_STATUS_NODE_PTR:
@@ -894,17 +894,25 @@ rec_get_nth_field_offs(
 				if SQL null; UNIV_SQL_DEFAULT is default value */
 {
 	ulint	offs;
-	ulint	length;
 	ut_ad(n < rec_offs_n_fields(offsets));
 	ut_ad(len);
 
 	if (n == 0) {
 		offs = 0;
 	} else {
-		offs = rec_offs_base(offsets)[n] & REC_OFFS_MASK;
+		offs = rec_offs_base(offsets)[n];
+		if (offs & REC_OFFS_DROP_COL
+		    || offs & REC_OFFS_DROP_SQL_NULL) {
+			ulint len;
+			offs  = offs & REC_OFFS_MASK;
+			rec_get_nth_field_offs(offsets, n-1, &len);
+			offs -= len;
+		} else {
+			offs = offs & REC_OFFS_MASK;
+		}
 	}
 
-	length = rec_offs_base(offsets)[1 + n];
+	ulint length = rec_offs_base(offsets)[1 + n];
 
 	if (length & REC_OFFS_SQL_NULL) {
 		length = UNIV_SQL_NULL;
@@ -1278,8 +1286,15 @@ rec_offs_data_size(
 	ulint	size;
 
 	ut_ad(rec_offs_validate(NULL, NULL, offsets));
-	size = rec_offs_base(offsets)[rec_offs_n_fields(offsets)]
-		& REC_OFFS_MASK;
+
+	ulint n = rec_offs_n_fields(offsets);
+
+	while (rec_offs_base(offsets)[n] & REC_OFFS_DROP_COL
+	       || rec_offs_base(offsets)[n] & REC_OFFS_DROP_SQL_NULL) {
+		n--;
+	}
+
+	size = rec_offs_base(offsets)[n] & REC_OFFS_MASK;
 	ut_ad(size < srv_page_size);
 	return(size);
 }
@@ -1426,11 +1441,19 @@ rec_get_converted_size(
 		      == DICT_FLD__SYS_INDEXES__MERGE_THRESHOLD);
 	} else {
 		ut_ad(dtuple->n_fields >= index->n_core_fields);
-		ut_ad(dtuple->n_fields <= index->n_fields);
+		ut_ad(dtuple->n_fields <= index->n_fields
+		      || dtuple->info_bits == REC_INFO_DEFAULT_ROW_DROP);
 	}
 #endif
 
 	if (dict_table_is_comp(index->table)) {
+
+		if (UNIV_UNLIKELY(dtuple_get_info_bits(dtuple)
+					== REC_INFO_DEFAULT_ROW_DROP)) {
+			return (rec_get_default_rec_converted_size(
+					index, dtuple, NULL));
+		}
+
 		return(rec_get_converted_size_comp(
 			       index,
 			       static_cast<rec_comp_status_t>(

storage/innobase/include/row0ins.h

@@ -136,7 +136,8 @@ dberr_t
 row_ins_index_entry_set_vals(
 	const dict_index_t*	index,
 	dtuple_t*		entry,
-	const dtuple_t*		row);
+	const dtuple_t*		row,
+	mem_heap_t*		heap);
 
 /***************************************************************//**
 Inserts an entry into a clustered index. Tries first optimistic,

storage/innobase/include/row0row.h

@@ -77,6 +77,19 @@ row_get_rec_roll_ptr(
 #define ROW_BUILD_FOR_PURGE	1	/*!< build row for purge. */
 #define ROW_BUILD_FOR_UNDO	2	/*!< build row for undo. */
 #define ROW_BUILD_FOR_INSERT	3	/*!< build row for insert. */
+
+void
+row_construct_default_row_blob(
+	dict_table_t*	table,
+	dfield_t*	dfield,
+	mem_heap_t*	heap);
+
+dtuple_t*
+row_build_clust_default_entry(
+        const dtuple_t*         row,
+        dict_index_t*           index,
+        mem_heap_t*             heap);
+
 /*****************************************************************//**
 When an insert or purge to a table is performed, this function builds
 the entry to be inserted into or purged from an index on the table.

storage/innobase/include/univ.i

@@ -220,6 +220,7 @@ and the insert buffer must be empty when the database is started */
 						adaptive hash index */
 #define UNIV_SRV_PRINT_LATCH_WAITS		/* enable diagnostic output
 						in sync0sync.cc */
+
 #define UNIV_BTR_PRINT				/* enable functions for
 						printing B-trees */
 #define UNIV_ZIP_DEBUG				/* extensive consistency checks

storage/innobase/page/page0cur.cc

@@ -1368,7 +1368,7 @@ page_cur_insert_rec_low(
 			switch (rec_get_status(current_rec)) {
 			case REC_STATUS_ORDINARY:
 			case REC_STATUS_NODE_PTR:
-			case REC_STATUS_COLUMNS_ADDED:
+			case REC_STATUS_COLUMNS_INSTANT:
 			case REC_STATUS_INFIMUM:
 				break;
 			case REC_STATUS_SUPREMUM:
@@ -1377,7 +1377,7 @@ page_cur_insert_rec_low(
 			switch (rec_get_status(insert_rec)) {
 			case REC_STATUS_ORDINARY:
 			case REC_STATUS_NODE_PTR:
-			case REC_STATUS_COLUMNS_ADDED:
+			case REC_STATUS_COLUMNS_INSTANT:
 				break;
 			case REC_STATUS_INFIMUM:
 			case REC_STATUS_SUPREMUM:

storage/innobase/page/page0page.cc

@@ -1826,6 +1826,7 @@ page_print_list(
 	count = 0;
 	for (;;) {
 		offsets = rec_get_offsets(cur.rec, index, offsets,
+					  page_rec_is_leaf(cur.rec),
 					  ULINT_UNDEFINED, &heap);
 		page_rec_print(cur.rec, offsets);
 
@@ -1848,6 +1849,7 @@ page_print_list(
 
 		if (count + pr_n >= n_recs) {
 			offsets = rec_get_offsets(cur.rec, index, offsets,
+						  page_rec_is_leaf(cur.rec),
 						  ULINT_UNDEFINED, &heap);
 			page_rec_print(cur.rec, offsets);
 		}

storage/innobase/row/row0ins.cc

@@ -2652,7 +2652,8 @@ row_ins_clust_index_entry_low(
 #endif /* UNIV_DEBUG */
 
 	if (UNIV_UNLIKELY(entry->info_bits != 0)) {
-		ut_ad(entry->info_bits == REC_INFO_DEFAULT_ROW);
+		ut_ad(entry->info_bits == REC_INFO_DEFAULT_ROW
+		      || entry->info_bits == REC_INFO_DEFAULT_ROW_DROP);
 		ut_ad(flags == BTR_NO_LOCKING_FLAG);
 		ut_ad(index->is_instant());
 		ut_ad(!dict_index_is_online_ddl(index));
@@ -3433,7 +3434,8 @@ dberr_t
 row_ins_index_entry_set_vals(
 	const dict_index_t*	index,
 	dtuple_t*		entry,
-	const dtuple_t*		row)
+	const dtuple_t*		row,
+	mem_heap_t*		heap)
 {
 	ulint	n_fields;
 	ulint	i;
@@ -3465,6 +3467,32 @@ row_ins_index_entry_set_vals(
 			ut_ad(dtuple_get_n_fields(row)
 			      == dict_table_get_n_cols(index->table));
 			row_field = dtuple_get_nth_v_field(row, v_col->v_pos);
+
+		} else if (col->is_dropped()) {
+
+			ut_ad(dict_index_is_clust(index));
+
+			field->type.prtype = DATA_NOT_NULL;
+
+			if (!(col->prtype & DATA_NOT_NULL)) {
+				field->data = 0x00;
+				field->len = UNIV_SQL_NULL;
+				field->type.prtype = DATA_BINARY_TYPE;
+			} else if (col->len > 0) {
+				field->data = mem_heap_zalloc(heap, col->len);
+				field->len = col->len;
+			} else {
+				field->data = 0x00;
+				field->len = 0;
+			}
+
+			if (col->len > 0) {
+				field->type.mtype = DATA_FIXBINARY;
+			} else {
+				field->type.mtype = DATA_BINARY;
+			}
+
+			continue;
 		} else {
 			row_field = dtuple_get_nth_field(
 				row, ind_field->col->ind);
@@ -3474,7 +3502,7 @@ row_ins_index_entry_set_vals(
 
 		/* Check column prefix indexes */
 		if (ind_field != NULL && ind_field->prefix_len > 0
-		    && dfield_get_len(row_field) != UNIV_SQL_NULL) {
+		    && len != UNIV_SQL_NULL) {
 
 			const	dict_col_t*	col
 				= dict_field_get_col(ind_field);
@@ -3528,7 +3556,8 @@ row_ins_index_entry_step(
 
 	ut_ad(dtuple_check_typed(node->row));
 
-	err = row_ins_index_entry_set_vals(node->index, node->entry, node->row);
+	err = row_ins_index_entry_set_vals(node->index, node->entry, node->row,
+					   node->entry_sys_heap);
 
 	if (err != DB_SUCCESS) {
 		DBUG_RETURN(err);

storage/innobase/row/row0log.cc

@@ -841,7 +841,7 @@ row_log_table_low_redundant(
 
 	const bool is_instant = index->online_log->is_instant(index);
 	rec_comp_status_t status = is_instant
-		? REC_STATUS_COLUMNS_ADDED : REC_STATUS_ORDINARY;
+		? REC_STATUS_COLUMNS_INSTANT : REC_STATUS_ORDINARY;
 
 	size = rec_get_converted_size_temp(
 		index, tuple->fields, tuple->n_fields, &extra_size, status);
@@ -895,7 +895,7 @@ row_log_table_low_redundant(
 			*b++ = static_cast<byte>(extra_size);
 		}
 
-		if (status == REC_STATUS_COLUMNS_ADDED) {
+		if (status == REC_STATUS_COLUMNS_INSTANT) {
 			ut_ad(is_instant);
 			if (n_fields <= index->online_log->n_core_fields) {
 				status = REC_STATUS_ORDINARY;
@@ -982,7 +982,7 @@ row_log_table_low(
 
 	ut_ad(page_is_comp(page_align(rec)));
 	ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY
-	      || rec_get_status(rec) == REC_STATUS_COLUMNS_ADDED);
+	      || rec_get_status(rec) == REC_STATUS_COLUMNS_INSTANT);
 
 	const ulint omit_size = REC_N_NEW_EXTRA_BYTES;
 

storage/innobase/row/row0purge.cc

@@ -838,8 +838,9 @@ static void row_purge_reset_trx_id(purge_node_t* node, mtr_t* mtr)
 		became purgeable) */
 		if (node->roll_ptr
 		    == row_get_rec_roll_ptr(rec, index, offsets)) {
-			ut_ad(!rec_get_deleted_flag(rec,
-						    rec_offs_comp(offsets)));
+			ut_ad(!rec_get_deleted_flag(
+					rec, rec_offs_comp(offsets))
+			      || rec_is_new_default_row(rec, index));
 			DBUG_LOG("purge", "reset DB_TRX_ID="
 				 << ib::hex(row_get_rec_trx_id(
 						    rec, index, offsets)));

storage/innobase/row/row0upd.cc

@@ -704,7 +704,7 @@ row_upd_rec_in_place(
 		switch (rec_get_status(rec)) {
 		case REC_STATUS_ORDINARY:
 			break;
-		case REC_STATUS_COLUMNS_ADDED:
+		case REC_STATUS_COLUMNS_INSTANT:
 			ut_ad(index->is_instant());
 			break;
 		case REC_STATUS_NODE_PTR:
@@ -1278,7 +1278,7 @@ row_upd_index_replace_new_col_val(
 	len = dfield_get_len(dfield);
 	data = static_cast<const byte*>(dfield_get_data(dfield));
 
-	if (field->prefix_len > 0) {
+	if (field && field->prefix_len > 0) {
 		ibool		fetch_ext = dfield_is_ext(dfield)
 			&& len < (ulint) field->prefix_len
 			+ BTR_EXTERN_FIELD_REF_SIZE;
@@ -1344,6 +1344,47 @@ row_upd_index_replace_new_col_val(
 	}
 }
 
+/** Apply an update vector to an default row entry.
+@param[in,out]	entry	index entry to be updated; the clustered index default
+			row record
+@param[in]	index	index of the entry
+@param[in]	update	update vector built for the entry
+@param[in,out]	heap	memory heap for copying off-page columns */
+static
+void
+row_upd_index_replace_default_rec_pos(
+	dtuple_t*		entry,
+	const dict_index_t*	index,
+	const upd_t*		update,
+	mem_heap_t*		heap)
+{
+	ut_ad(!index->table->skip_alter_undo);
+
+	const page_size_t&	page_size = dict_table_page_size(index->table);
+
+	dtuple_set_info_bits(entry, update->info_bits);
+
+	for (unsigned i = entry->n_fields;
+	     i >= unsigned(index->n_uniq + DATA_ROLL_PTR); i--) {
+		const dict_field_t*	field = NULL;
+		const dict_col_t*	col = NULL;
+		const upd_field_t*	uf;
+
+		uf = upd_get_field_by_field_no(update, i -1, false);
+
+		if (uf) {
+			if (i > unsigned(index->n_uniq + DATA_ROLL_PTR)) {
+				field = dict_index_get_nth_field(index, i - 1);
+				col = dict_field_get_col(field);
+			}
+
+			row_upd_index_replace_new_col_val(
+				dtuple_get_nth_field(entry, i),
+				field, col, uf, heap, page_size);
+		}
+	}
+}
+
 /** Apply an update vector to an index entry.
 @param[in,out]	entry	index entry to be updated; the clustered index record
 			must be covered by a lock or a page latch to prevent
@@ -1364,6 +1405,11 @@ row_upd_index_replace_new_col_vals_index_pos(
 
 	dtuple_set_info_bits(entry, update->info_bits);
 
+	if (UNIV_UNLIKELY(entry->info_bits == REC_INFO_DEFAULT_ROW_DROP)) {
+		row_upd_index_replace_default_rec_pos(entry, index, update, heap);
+		return;
+	}
+
 	for (unsigned i = index->n_fields; i--; ) {
 		const dict_field_t*	field;
 		const dict_col_t*	col;

storage/innobase/trx/trx0rec.cc

@@ -506,7 +506,8 @@ trx_undo_page_report_insert(
 	/* Store then the fields required to uniquely determine the record
 	to be inserted in the clustered index */
 	if (UNIV_UNLIKELY(clust_entry->info_bits != 0)) {
-		ut_ad(clust_entry->info_bits == REC_INFO_DEFAULT_ROW);
+		ut_ad(clust_entry->info_bits == REC_INFO_DEFAULT_ROW
+		      || clust_entry->info_bits == REC_INFO_DEFAULT_ROW_DROP);
 		ut_ad(index->is_instant());
 		ut_ad(undo_block->frame[first_free + 2]
 		      == TRX_UNDO_INSERT_REC);
@@ -922,7 +923,8 @@ trx_undo_page_report_modify(
 	if (!update) {
 		ut_ad(!rec_get_deleted_flag(rec, dict_table_is_comp(table)));
 		type_cmpl = TRX_UNDO_DEL_MARK_REC;
-	} else if (rec_get_deleted_flag(rec, dict_table_is_comp(table))) {
+	} else if (rec_get_deleted_flag(rec, dict_table_is_comp(table))
+		   && !rec_is_new_default_row(rec, index)) {
 		/* In delete-marked records, DB_TRX_ID must
 		always refer to an existing update_undo log record. */
 		ut_ad(row_get_rec_trx_id(rec, index, offsets));
@@ -1094,8 +1096,14 @@ trx_undo_page_report_modify(
 						flen, max_v_log_len);
 				}
 			} else {
-				field = rec_get_nth_cfield(
-					rec, index, offsets, pos, &flen);
+				if (UNIV_UNLIKELY(rec_is_new_default_row(
+							rec, index))) {
+					field = rec_get_nth_def_field(
+						rec, index, offsets, pos, &flen);
+				} else {
+					field = rec_get_nth_cfield(
+						rec, index, offsets, pos, &flen);
+				}
 			}
 
 			if (trx_undo_left(undo_block, ptr) < 15) {