Merge branch 'mdev-18188' of https://github.com/tliu22/server into 10.5bb-10.5-MDEV-18188

16 files changed, 686 insertions, 51 deletions

diff --git a/mysql-test/suite/innodb/r/read_only_recover_committed.result b/mysql-test/suite/innodb/r/read_only_recover_committed.result
index e7895ce6d46..d9358df74e5 100644
--- a/mysql-test/suite/innodb/r/read_only_recover_committed.result
+++ b/mysql-test/suite/innodb/r/read_only_recover_committed.result
@@ -65,7 +65,6 @@ SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
 SELECT * FROM t;
 a
 3
-20
 # restart
 SELECT * FROM t;
 a
diff --git a/sql/handler.h b/sql/handler.h
index 41da458aa2a..c0cf1f1abbb 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -3632,6 +3632,12 @@ public:
   */
   virtual int pre_records() { return 0; }
   virtual ha_rows records() { return stats.records; }
+  virtual int records2(ha_rows* num_rows)
+  {
+    *num_rows = records();
+    return 0;
+  }
+
   /**
     Return upper bound of current number of records in the table
     (max. of how many records one will retrieve when doing a full table scan)
diff --git a/sql/opt_sum.cc b/sql/opt_sum.cc
index af2d9ddc2e7..6cf09afad9d 100644
--- a/sql/opt_sum.cc
+++ b/sql/opt_sum.cc
@@ -298,6 +298,7 @@ int opt_sum_query(THD *thd,
     if (!(tl->table->file->ha_table_flags() & HA_STATS_RECORDS_IS_EXACT) ||
         tl->schema_table)
     {
+stats_records_inexact:
       maybe_exact_count&= MY_TEST(!tl->schema_table &&
                                   (tl->table->file->ha_table_flags() &
                                    HA_HAS_RECORDS));
@@ -321,7 +322,12 @@ int opt_sum_query(THD *thd,
         tl->table->file->print_error(error, MYF(ME_FATAL));
         DBUG_RETURN(error);
       }
-      count*= tl->table->file->stats.records;
+
+      ha_rows num_rows;
+      if (tl->table->file->records2(&num_rows)) {
+        goto stats_records_inexact;
+      }
+      count *= num_rows;
     }
   }
 
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index 846d8ecfd7e..0286ba1996a 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -578,7 +578,7 @@ incompatible:
 			}
 		}
 
-		if (index->table->deserialise_columns(
+		if (index->table->deserialise_mblob(
 			    &block->frame[FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE],
 			    len)) {
 			goto incompatible;
@@ -4391,7 +4391,7 @@ static void btr_cur_trim_alter_metadata(dtuple_t* entry,
 		return;
 	}
 
-	/* This is based on dict_table_t::deserialise_columns()
+	/* This is based on dict_table_t::deserialise_mblob()
 	and btr_cur_instant_init_low(). */
 	mtr_t mtr;
 	mtr.start();
diff --git a/storage/innobase/dict/dict0load.cc b/storage/innobase/dict/dict0load.cc
index 753bcf74967..64f25685451 100644
--- a/storage/innobase/dict/dict0load.cc
+++ b/storage/innobase/dict/dict0load.cc
@@ -1236,7 +1236,7 @@ dict_sys_tables_rec_read(
 	ATOMIC_WRITES=OFF, but it would be ignored starting with
 	MariaDB 10.2.4. */
 	compile_time_assert(DICT_TF_POS_PAGE_COMPRESSION == 7);
-	compile_time_assert(DICT_TF_POS_UNUSED == 14);
+	compile_time_assert(DICT_TF_POS_UNUSED == 15);
 
 	if ((type & 0x19f) != 0x101) {
 		/* The table cannot have been created with MariaDB
diff --git a/storage/innobase/dict/dict0mem.cc b/storage/innobase/dict/dict0mem.cc
index 96f2d7b6e3b..912a6a82af7 100644
--- a/storage/innobase/dict/dict0mem.cc
+++ b/storage/innobase/dict/dict0mem.cc
@@ -1263,16 +1263,22 @@ inline void dict_index_t::reconstruct_fields()
 	n_core_null_bytes = static_cast<byte>(UT_BITS_IN_BYTES(n_core_null));
 }
 
-/** Reconstruct dropped or reordered columns.
-@param[in]	metadata	data from serialise_columns()
+/** Deserialise metadata BLOB and reconstruct dropped or reordered columns,
+committed count.
+@param[in]	metadata	data from serialise_mblob()
 @param[in]	len		length of the metadata, in bytes
 @return whether parsing the metadata failed */
-bool dict_table_t::deserialise_columns(const byte* metadata, ulint len)
+bool dict_table_t::deserialise_mblob(const byte* metadata, ulint len)
 {
 	DBUG_ASSERT(!instant);
 
 	unsigned num_non_pk_fields = mach_read_from_4(metadata);
-	metadata += 4;
+	metadata += NUM_NON_PK_FIELDS_SIZE;
+
+	DBUG_ASSERT((NUM_NON_PK_FIELDS_SIZE + num_non_pk_fields * NON_PK_FIELD_SIZE
+			== len)
+		|| (NUM_NON_PK_FIELDS_SIZE + num_non_pk_fields * NON_PK_FIELD_SIZE
+			+ COMMITTED_COUNT_SIZE == len));
 
 	if (num_non_pk_fields >= REC_MAX_N_FIELDS - 3) {
 		return true;
@@ -1293,7 +1299,7 @@ bool dict_table_t::deserialise_columns(const byte* metadata, ulint len)
 
 	for (unsigned i = 0; i < num_non_pk_fields; i++) {
 		auto c = field_map[i] = mach_read_from_2(metadata);
-		metadata += 2;
+		metadata += NON_PK_FIELD_SIZE;
 
 		if (field_map[i].is_dropped()) {
 			if (c.ind() > DICT_MAX_FIXED_COL_LEN + 1) {
@@ -1326,6 +1332,14 @@ bool dict_table_t::deserialise_columns(const byte* metadata, ulint len)
 	DBUG_ASSERT(col == &dropped_cols[n_dropped_cols]);
 
 	UT_LIST_GET_FIRST(indexes)->reconstruct_fields();
+
+	committed_count_inited = (NUM_NON_PK_FIELDS_SIZE +
+		num_non_pk_fields * NON_PK_FIELD_SIZE + COMMITTED_COUNT_SIZE == len);
+	if (committed_count_inited) {
+		committed_count = mach_read_from_8(metadata);
+		metadata += COMMITTED_COUNT_SIZE;
+	}
+
 	return false;
 }
 
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index a35b90e91f0..827b63ef559 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -622,6 +622,11 @@ static MYSQL_THDVAR_BOOL(compression_default, PLUGIN_VAR_OPCMDARG,
   "Is compression the default for new tables", 
   NULL, NULL, FALSE);
 
+// should persistent count be enabled by default for new tables
+static MYSQL_THDVAR_BOOL(persistent_count_default, PLUGIN_VAR_OPCMDARG,
+  "Is persistent count enabled by default for new tables",
+  NULL, NULL, FALSE);
+
 /** Update callback for SET [SESSION] innodb_default_encryption_key_id */
 static void
 innodb_default_encryption_key_id_update(THD* thd, st_mysql_sys_var* var,
@@ -663,7 +668,7 @@ ha_create_table_option innodb_table_option_list[]=
   HA_TOPTION_ENUM("ENCRYPTED", encryption, "DEFAULT,YES,NO", 0),
   /* With this option the user defines the key identifier using for the encryption */
   HA_TOPTION_SYSVAR("ENCRYPTION_KEY_ID", encryption_key_id, default_encryption_key_id),
-
+  HA_TOPTION_SYSVAR("PERSISTENT_COUNT", persistent_count, persistent_count_default),
   HA_TOPTION_END
 };
 
@@ -4763,6 +4768,9 @@ ha_innobase::table_flags() const
 	called before prebuilt is inited. */
 
 	if (thd_tx_isolation(thd) <= ISO_READ_COMMITTED) {
+		if (thd_tx_isolation(thd) == ISO_READ_COMMITTED) {
+			return (flags | HA_STATS_RECORDS_IS_EXACT);
+		}
 		return(flags);
 	}
 
@@ -11478,8 +11486,9 @@ index_bad:
 	dict_tf_set(&m_flags, innodb_row_format, zip_ssize,
 			m_use_data_dir,
 			options->page_compressed,
-		    	options->page_compression_level == 0 ?
-		        default_compression_level : ulint(options->page_compression_level));
+			options->page_compression_level == 0 ? default_compression_level
+				: ulint(options->page_compression_level),
+			options->persistent_count);
 
 	if (m_form->s->table_type == TABLE_TYPE_SEQUENCE) {
 		m_flags |= DICT_TF_MASK_NO_ROLLBACK;
@@ -12968,6 +12977,15 @@ ha_innobase::create(
 		DBUG_RETURN(error);
 	}
 
+	dict_table_t* ib_table = info.table();
+
+	ib_table->committed_count_inited =
+		!!DICT_TF_GET_PERSISTENT_COUNT(info.flags());
+	if (ib_table->committed_count_inited) {
+		ib_table->committed_count = 0;
+		innobase_create_persistent_count(ib_table, form, trx);
+	}
+
 	innobase_commit_low(trx);
 	row_mysql_unlock_data_dictionary(trx);
 
@@ -13712,6 +13730,65 @@ ha_innobase::rename_table(
 	DBUG_RETURN(convert_error_code_to_mysql(error, 0, NULL));
 }
 
+/** Initialize committed count within dict_table_t.
+@return  0 or error code */
+int
+ha_innobase::enable_persistent_count()
+{
+	dict_table_t* ib_table = m_prebuilt->table;
+	uchar* read_buf = m_prebuilt->m_mysql_table->record[0];
+	trx_t* trx = m_prebuilt->trx;
+	int err;
+
+	if (ib_table->committed_count_inited) {
+		return -1;  /* Already initialized */
+	}
+
+	uint64_t count = 0;
+	rnd_init(true);
+	do {
+		err = rnd_next(read_buf);
+		if (!err) {
+			count++;
+		}
+	} while (!err);
+	ib_table->committed_count = count - trx->uncommitted_count(ib_table);
+	ib_table->committed_count_inited = true;
+
+	return 0;
+}
+
+/** De-initialize committed count within dict_table_t.
+@return  0 or error code */
+int
+ha_innobase::disable_persistent_count()
+{
+	dict_table_t* ib_table = m_prebuilt->table;
+	ib_table->committed_count_inited = false;
+
+	return 0;
+}
+
+/** If committed count is initialized and transaction is in READ COMMITTED mode,
+returns exact number of records; otherwise returns an estimate of index records.
+@param[out]  num_rows  number of rows in table
+@return  0 if success
+@return  1 if failure */
+inline
+int
+ha_innobase::records2(ha_rows* num_rows)
+{	
+    trx_t* trx = m_prebuilt->trx;
+    dict_table_t* ib_table = m_prebuilt->table;
+
+    if (!ib_table->committed_count_inited) {
+        return 1;
+    }
+    *num_rows = ib_table->committed_count + trx->uncommitted_count(ib_table);
+
+    return 0;
+}
+
 /*********************************************************************//**
 Estimates the number of index records in a range.
 @return estimated number of rows */
@@ -20143,6 +20220,7 @@ static struct st_mysql_sys_var* innobase_system_variables[]= {
   MYSQL_SYSVAR(buf_dump_status_frequency),
   MYSQL_SYSVAR(background_thread),
   MYSQL_SYSVAR(encrypt_temporary_tables),
+  MYSQL_SYSVAR(persistent_count_default),
 
   NULL
 };
diff --git a/storage/innobase/handler/ha_innodb.h b/storage/innobase/handler/ha_innodb.h
index c5c5e8cba31..c8c21a6cec3 100644
--- a/storage/innobase/handler/ha_innodb.h
+++ b/storage/innobase/handler/ha_innodb.h
@@ -50,6 +50,7 @@ struct ha_table_option_struct
 						value OFF.*/
 	uint		encryption;		/*!<  DEFAULT, ON, OFF */
 	ulonglong	encryption_key_id;	/*!< encryption key id  */
+	bool		persistent_count;	/*!< persistent count */
 };
 
 /** The class defining a handle to an Innodb table */
@@ -184,6 +185,11 @@ public:
                 const key_range*        max_key,
                 page_range*             pages) override;
 
+	int enable_persistent_count();
+	int disable_persistent_count();
+
+	inline int records2(ha_rows* num_rows) override;
+
 	ha_rows estimate_rows_upper_bound() override;
 
 	void update_create_info(HA_CREATE_INFO* create_info) override;
@@ -721,6 +727,10 @@ public:
 	THD* thd() const
 	{ return(m_thd); }
 
+	/** Get table. */
+	dict_table_t* table() const
+	{ return(m_table); }
+
 	/** Normalizes a table name string.
 	A normalized name consists of the database name catenated to '/' and
 	table name. An example: test/mytable. On Windows normalization puts
@@ -955,6 +965,26 @@ ib_push_frm_error(
 	ulint		n_keys,		/*!< in: InnoDB #keys */
 	bool		push_warning);	/*!< in: print warning ? */
 
+/** Initialize metadata BLOB on CREATE for persistent count.
+@param[in]  user_table  InnoDB table
+@param[in]  table       MySQL table
+@param[in]  trx         transaction
+@retval true   failure
+@retval false  success */
+bool innobase_create_persistent_count(
+	dict_table_t*       user_table,
+	const TABLE*        table,
+	trx_t*              trx);
+
+/** Update metadata BLOB to reflect updated persistent count.
+@param[in]  user_table  InnoDB table
+@param[in]	trx		transaction
+@retval	true	failure
+@retval	false	success */
+bool innobase_update_persistent_count(
+	const dict_table_t* user_table,
+	trx_t*		trx);
+
 /** Check each index part length whether they not exceed the max limit
 @param[in]	max_field_len	maximum allowed key part length
 @param[in]	key		MariaDB key definition
diff --git a/storage/innobase/handler/handler0alter.cc b/storage/innobase/handler/handler0alter.cc
index 710a3787e09..8159d1be815 100644
--- a/storage/innobase/handler/handler0alter.cc
+++ b/storage/innobase/handler/handler0alter.cc
@@ -208,7 +208,8 @@ inline void dict_table_t::init_instant(const dict_table_t& table)
 @param[out]	first_alter_pos	0, or 1 + first changed column position */
 inline void dict_table_t::prepare_instant(const dict_table_t& old,
 					  const ulint* col_map,
-					  unsigned& first_alter_pos)
+					  unsigned& first_alter_pos,
+					  bool alter_persistent_count)
 {
 	DBUG_ASSERT(!is_instant());
 	DBUG_ASSERT(n_dropped() == 0);
@@ -230,6 +231,10 @@ inline void dict_table_t::prepare_instant(const dict_table_t& old,
 	dict_index_t& index = *indexes.start;
 	first_alter_pos = 0;
 
+	if (alter_persistent_count) {
+		goto add_metadata;
+	}
+
 	for (unsigned i = 0; i + DATA_N_SYS_COLS < old.n_cols; i++) {
 		if (col_map[i] != i) {
 			first_alter_pos = 1 + i;
@@ -922,6 +927,9 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 	/** The page_compression_level attribute, or 0 */
 	const uint	page_compression_level;
 
+	/** Enable persistent count */
+	const bool persistent_count;
+
 	ha_innobase_inplace_ctx(row_prebuilt_t*& prebuilt_arg,
 				dict_index_t** drop_arg,
 				ulint num_to_drop_arg,
@@ -938,7 +946,8 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 				ulonglong autoinc_col_max_value_arg,
 				bool allow_not_null_flag,
 				bool page_compressed,
-				ulonglong page_compression_level_arg) :
+				ulonglong page_compression_level_arg,
+				bool persistent_count) :
 		inplace_alter_handler_ctx(),
 		prebuilt (prebuilt_arg),
 		add_index (0), add_key_numbers (0), num_to_add_index (0),
@@ -979,7 +988,8 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 				       ? (page_compression_level_arg
 					  ? uint(page_compression_level_arg)
 					  : page_zip_level)
-				       : 0)
+				       : 0),
+		persistent_count(persistent_count)
 	{
 		ut_ad(old_n_cols >= DATA_N_SYS_COLS);
 		ut_ad(page_compression_level <= 9);
@@ -1038,10 +1048,13 @@ struct ha_innobase_inplace_ctx : public inplace_alter_handler_ctx
 
 		instant_table = new_table;
 		new_table = old_table;
-		export_vars.innodb_instant_alter_column++;
+
+		if  (!persistent_count) {
+			export_vars.innodb_instant_alter_column++;
+		}
 
 		instant_table->prepare_instant(*old_table, col_map,
-					       first_alter_pos);
+					       first_alter_pos, persistent_count);
 	}
 
 	/** Adjust table metadata for instant ADD/DROP/reorder COLUMN.
@@ -1313,9 +1326,10 @@ static bool alter_options_need_rebuild(
 
 	/* Allow an instant change to enable page_compressed,
 	and any change of page_compression_level. */
-	if ((!alt_opt.page_compressed && opt.page_compressed)
-	    || alt_opt.encryption != opt.encryption
-	    || alt_opt.encryption_key_id != opt.encryption_key_id) {
+	if (((!alt_opt.page_compressed && opt.page_compressed)
+	     || alt_opt.encryption != opt.encryption
+	     || alt_opt.encryption_key_id != opt.encryption_key_id)
+	    || alt_opt.persistent_count != opt.persistent_count) {
 		return(true);
 	}
 
@@ -1437,7 +1451,7 @@ check_v_col_in_order(
 	return(true);
 }
 
-/** Determine if an instant operation is possible for altering columns.
+/** Determine if an instant operation is possible for an ALTER.
 @param[in]	ib_table	InnoDB table definition
 @param[in]	ha_alter_info	the ALTER TABLE operation
 @param[in]	table		table definition before ALTER TABLE
@@ -1445,7 +1459,7 @@ check_v_col_in_order(
 @param[in]	strict		whether to ensure that user records fit */
 static
 bool
-instant_alter_column_possible(
+instant_alter_possible(
 	const dict_table_t&		ib_table,
 	const Alter_inplace_info*	ha_alter_info,
 	const TABLE*			table,
@@ -1456,6 +1470,11 @@ instant_alter_column_possible(
 	ut_ad(pk->is_primary());
 	ut_ad(!pk->has_virtual());
 
+	if (table->s->option_struct->persistent_count
+		!= altered_table->s->option_struct->persistent_count) {
+		return true;
+	}
+
 	if (ha_alter_info->handler_flags
 	    & (ALTER_STORED_COLUMN_ORDER | ALTER_DROP_STORED_COLUMN
 	       | ALTER_ADD_STORED_BASE_COLUMN)) {
@@ -2397,7 +2416,7 @@ next_column:
 		af++;
 	}
 
-	const bool supports_instant = instant_alter_column_possible(
+	const bool supports_instant = instant_alter_possible(
 		*m_prebuilt->table, ha_alter_info, table, altered_table,
 		is_innodb_strict_mode());
 	if (add_drop_v_cols) {
@@ -5502,18 +5521,22 @@ innobase_drop_virtual_try(
 	return false;
 }
 
-/** Serialise metadata of dropped or reordered columns.
+/** Serialise metadata BLOB, consisting of dropped or reordered columns,
+committed count.
 @param[in,out]	heap	memory heap for allocation
 @param[out]	field	data field with the metadata */
 inline
-void dict_table_t::serialise_columns(mem_heap_t* heap, dfield_t* field) const
+void dict_table_t::serialise_mblob(mem_heap_t* heap, dfield_t* field) const
 {
 	DBUG_ASSERT(instant);
-	const dict_index_t& index = *UT_LIST_GET_FIRST(indexes);
+	dict_index_t& index = *UT_LIST_GET_FIRST(indexes);
 	unsigned n_fixed = index.first_user_field();
 	unsigned num_non_pk_fields = index.n_fields - n_fixed;
 
-	ulint len = 4 + num_non_pk_fields * 2;
+	ulint len = NUM_NON_PK_FIELDS_SIZE + num_non_pk_fields * NON_PK_FIELD_SIZE;
+	if (committed_count_inited) {
+		len += COMMITTED_COUNT_SIZE;
+	}
 
 	byte* data = static_cast<byte*>(mem_heap_alloc(heap, len));
 
@@ -5521,11 +5544,16 @@ void dict_table_t::serialise_columns(mem_heap_t* heap, dfield_t* field) const
 
 	mach_write_to_4(data, num_non_pk_fields);
 
-	data += 4;
+	data += NUM_NON_PK_FIELDS_SIZE;
 
 	for (ulint i = n_fixed; i < index.n_fields; i++) {
 		mach_write_to_2(data, instant->field_map[i - n_fixed]);
-		data += 2;
+		data += NON_PK_FIELD_SIZE;
+	}
+
+	if (committed_count_inited) {
+		mach_write_to_8(data, committed_count);
+		data += COMMITTED_COUNT_SIZE;
 	}
 }
 
@@ -5556,7 +5584,7 @@ dict_index_t::instant_metadata(const dtuple_t& row, mem_heap_t* heap) const
 		dfield_t* dfield = dtuple_get_nth_field(entry, i);
 
 		if (i == first_user_field()) {
-			table->serialise_columns(heap, dfield);
+			table->serialise_mblob(heap, dfield);
 			dfield->type.metadata_blob_init();
 			field--;
 			continue;
@@ -5770,7 +5798,7 @@ add_all_virtual:
 		   && innobase_add_virtual_try(ha_alter_info, user_table,
 					       trx)) {
 		return true;
-        }
+    }
 
 	if (!user_table->space) {
 		/* In case of ALTER TABLE...DISCARD TABLESPACE,
@@ -5959,6 +5987,286 @@ func_exit:
 	return false;
 }
 
+/** Initialize metadata BLOB on CREATE for persistent count. */
+bool innobase_create_persistent_count(
+	dict_table_t* user_table,
+	const TABLE*  table,
+	trx_t*		  trx)
+{
+	mem_heap_t* heap = mem_heap_create(1024);
+	dict_index_t* index = dict_table_get_first_index(user_table);
+
+	user_table->instant = new (mem_heap_alloc(user_table->heap,
+	                            sizeof(dict_instant_t))) dict_instant_t();
+	user_table->instant->n_dropped = 0;
+	user_table->instant->dropped = NULL;
+
+	const unsigned u = index->first_user_field();
+	field_map_element_t* field_map_it = static_cast<field_map_element_t*>(
+		mem_heap_zalloc(user_table->heap, (index->n_fields - u)
+						* sizeof *field_map_it));
+	user_table->instant->field_map = field_map_it;
+	for (unsigned i = u; i < index->n_fields; i++) {
+		dict_field_t& f = index->fields[i];
+		ut_ad(!f.col->is_dropped());
+		(*field_map_it++).set_ind(f.col->ind);
+	}
+
+	mtr_t mtr;
+	/* Construct a table row of default values for the stored columns. */
+	dtuple_t* row = dtuple_create(heap, user_table->n_cols);
+	dict_table_copy_types(row, user_table);
+	Field** af = table->field;
+	Field** const end = table->field + table->s->fields;
+
+	for (uint i = 0; af < end; af++) {
+		if (!(*af)->stored_in_db()) {
+			continue;
+		}
+
+		dfield_t* d = dtuple_get_nth_field(row, i);
+		const dict_col_t* col = dict_table_get_nth_col(user_table, i);
+		DBUG_ASSERT(!col->is_virtual());
+		DBUG_ASSERT(!col->is_dropped());
+		DBUG_ASSERT(col->mtype != DATA_SYS);
+		DBUG_ASSERT(!strcmp((*af)->field_name.str,
+				    dict_table_get_col_name(user_table, i)));
+
+		if (col->is_added()) {
+			dfield_set_data(d, col->def_val.data,
+					col->def_val.len);
+		} else if ((*af)->real_maybe_null()) {
+			/* Store NULL for nullable 'core' columns. */
+			dfield_set_null(d);
+		} else {
+			switch ((*af)->type()) {
+			case MYSQL_TYPE_VARCHAR:
+			case MYSQL_TYPE_GEOMETRY:
+			case MYSQL_TYPE_TINY_BLOB:
+			case MYSQL_TYPE_MEDIUM_BLOB:
+			case MYSQL_TYPE_BLOB:
+			case MYSQL_TYPE_LONG_BLOB:
+				/* Store the empty string for 'core'
+				variable-length NOT NULL columns. */
+				dfield_set_data(d, field_ref_zero, 0);
+				break;
+			default:
+				/* For fixed-length NOT NULL 'core' columns,
+				get a dummy default value from SQL. Note that
+				we will preserve the old values of these
+				columns when updating the metadata
+				record, to avoid unnecessary updates. */
+				ulint len = (*af)->pack_length();
+				DBUG_ASSERT(d->type.mtype != DATA_INT
+					    || len <= 8);
+				row_mysql_store_col_in_innobase_format(
+					d, d->type.mtype == DATA_INT
+					? static_cast<byte*>(
+						mem_heap_alloc(heap, len))
+					: NULL, true, (*af)->ptr, len,
+					dict_table_is_comp(user_table));
+			}
+		}
+
+		i++;
+	}
+
+	unsigned i = unsigned(user_table->n_cols) - DATA_N_SYS_COLS;
+	DBUG_ASSERT(i >= table->s->stored_fields);
+	DBUG_ASSERT(i <= table->s->stored_fields + 1);
+	if (i > table->s->fields) {
+		const dict_col_t& fts_doc_id = user_table->cols[i - 1];
+		DBUG_ASSERT(!strcmp(fts_doc_id.name(*user_table),
+				    FTS_DOC_ID_COL_NAME));
+		DBUG_ASSERT(!fts_doc_id.is_nullable());
+		DBUG_ASSERT(fts_doc_id.len == 8);
+		dfield_set_data(dtuple_get_nth_field(row, i - 1),
+				field_ref_zero, fts_doc_id.len);
+	}
+	byte trx_id[DATA_TRX_ID_LEN], roll_ptr[DATA_ROLL_PTR_LEN];
+	dfield_set_data(dtuple_get_nth_field(row, i++), field_ref_zero,
+			DATA_ROW_ID_LEN);
+	dfield_set_data(dtuple_get_nth_field(row, i++), trx_id, sizeof trx_id);
+	dfield_set_data(dtuple_get_nth_field(row, i),roll_ptr,sizeof roll_ptr);
+	DBUG_ASSERT(i + 1 == user_table->n_cols);
+
+	trx_write_trx_id(trx_id, trx->id);
+	/* The DB_ROLL_PTR will be assigned later, when allocating undo log.
+	Silence a Valgrind warning in dtuple_validate() when
+	row_ins_clust_index_entry_low() searches for the insert position. */
+	memset(roll_ptr, 0, sizeof roll_ptr);
+
+	dtuple_t* entry = index->instant_metadata(*row, heap);
+	mtr.start();
+	index->set_modified(mtr);
+	btr_pcur_t pcur;
+	btr_pcur_open_at_index_side(true, index, BTR_MODIFY_TREE, &pcur, true,
+				    0, &mtr);
+	ut_ad(btr_pcur_is_before_first_on_page(&pcur));
+	btr_pcur_move_to_next_on_page(&pcur);
+
+	buf_block_t* block = btr_pcur_get_block(&pcur);
+	ut_ad(page_is_leaf(block->frame));
+	ut_ad(!page_has_prev(block->frame));
+	ut_ad(!buf_block_get_page_zip(block));
+	que_thr_t* thr = pars_complete_graph_for_exec(NULL, trx, heap, NULL);
+
+	dberr_t err = DB_SUCCESS;
+	/* Convert the table to the instant ALTER TABLE format. */
+	mtr.commit();
+	mtr.start();
+	index->set_modified(mtr);
+	if (buf_block_t* root = btr_root_block_get(index, RW_SX_LATCH, &mtr)) {
+		if (fil_page_get_type(root->frame) != FIL_PAGE_INDEX) {
+			DBUG_ASSERT(!"wrong page type");
+			goto err_exit;
+		}
+
+		btr_set_instant(root, *index, &mtr);
+		mtr.commit();
+		mtr.start();
+		index->set_modified(mtr);
+		err = row_ins_clust_index_entry_low(
+			BTR_NO_LOCKING_FLAG, BTR_MODIFY_TREE, index,
+			index->n_uniq, entry, 0, thr);
+	} else {
+err_exit:
+		err = DB_CORRUPTION;
+	}
+
+	mtr.commit();
+	mem_heap_free(heap);
+
+	if (err != DB_SUCCESS) {
+		my_error_innodb(err, table->s->table_name.str,
+				user_table->flags);
+		return true;
+	}
+
+	return false;
+}
+
+/** Update metadata BLOB to reflect updated persistent count.
+@param[in]  user_table  InnoDB table
+@param[in]	trx		transaction
+@retval	true	failure
+@retval	false	success */
+bool innobase_update_persistent_count(
+	const dict_table_t* user_table,
+	trx_t*		  trx)
+{
+	mem_heap_t* heap = mem_heap_create(1024);
+	dict_index_t* index = dict_table_get_first_index(user_table);
+	mtr_t mtr;
+	const unsigned n_old_fields = index->n_fields;
+
+	mutex_enter(&dict_sys.mutex);
+
+	mtr.start();
+	index->set_modified(mtr);
+	btr_pcur_t pcur;
+	btr_pcur_open_at_index_side(true, index, BTR_MODIFY_TREE, &pcur, true,
+				    0, &mtr);
+	ut_ad(btr_pcur_is_before_first_on_page(&pcur));
+	btr_pcur_move_to_next_on_page(&pcur);
+
+	buf_block_t* block = btr_pcur_get_block(&pcur);
+	ut_ad(page_is_leaf(block->frame));
+	ut_ad(!page_has_prev(block->frame));
+	ut_ad(!buf_block_get_page_zip(block));
+	const rec_t* rec = btr_pcur_get_rec(&pcur);
+	que_thr_t* thr = pars_complete_graph_for_exec(NULL, trx, heap, NULL);
+	dberr_t err = DB_SUCCESS;
+
+	ut_ad(rec_is_metadata(rec, *index));
+	ut_ad(page_rec_is_user_rec(rec));
+
+	/* Ensure that the root page is in the correct format. */
+	buf_block_t* root = btr_root_block_get(index, RW_X_LATCH,
+							&mtr);
+	DBUG_ASSERT(root);
+	if (fil_page_get_type(root->frame) != FIL_PAGE_TYPE_INSTANT) {
+		DBUG_ASSERT(!"wrong page type");
+		err = DB_CORRUPTION;
+		goto func_exit;
+	}
+	{
+		mem_heap_t* offsets_heap = NULL;
+		rec_offs* offsets = rec_get_offsets(rec, index, NULL, true,
+							ULINT_UNDEFINED, &offsets_heap);
+		dtuple_t* entry = row_metadata_to_tuple(rec, index, offsets,
+							heap, REC_INFO_METADATA_ALTER, !trx->in_rollback);
+		dfield_t* dfield = dtuple_get_nth_field(entry,
+								index->first_user_field());
+		// ASSERT for mblob_init ?
+		user_table->serialise_mblob(heap, dfield);
+
+		btr_set_instant(root, *index, &mtr);
+
+		/* Extend the record with any added columns. */
+		uint n = uint(index->n_fields) - n_old_fields;
+		/* Reserve room for DB_TRX_ID,DB_ROLL_PTR and any
+		non-updated off-page columns in case they are moved off
+		page as a result of the update. */
+		ut_ad(user_table->instant);
+		upd_t* update = upd_create(index->n_fields + 1, heap);
+		update->n_fields = n + 1;
+		update->info_bits = REC_INFO_METADATA_ALTER;
+		upd_field_t* uf = upd_get_nth_field(update, 0);
+		uf->field_no = index->first_user_field();
+		uf->new_val = entry->fields[uf->field_no];
+		DBUG_ASSERT(!dfield_is_ext(&uf->new_val));
+		DBUG_ASSERT(!dfield_is_null(&uf->new_val));
+
+		/* Add the default values for instantly added columns */
+		unsigned j = 1;
+
+		for (unsigned k = n_old_fields; k < index->n_fields; k++) {
+			upd_field_t* uf = upd_get_nth_field(update, j++);
+			uf->field_no = static_cast<uint16_t> (k + 1);
+			uf->new_val = entry->fields[k + 1];
+
+			ut_ad(j <= n + 1);
+		}
+
+		ut_ad(j == n + 1);
+
+		offsets = NULL;
+		offsets_heap = NULL;
+		big_rec_t* big_rec;
+		err = btr_cur_pessimistic_update(
+			BTR_NO_LOCKING_FLAG | BTR_KEEP_POS_FLAG,
+			btr_pcur_get_btr_cur(&pcur),
+			&offsets, &offsets_heap, heap,
+			&big_rec, update, UPD_NODE_NO_ORD_CHANGE,
+			thr, trx->id, &mtr);
+
+		offsets = rec_get_offsets(
+			btr_pcur_get_rec(&pcur), index, offsets,
+			true, ULINT_UNDEFINED, &offsets_heap);
+		if (big_rec) {
+			if (err == DB_SUCCESS) {
+				err = btr_store_big_rec_extern_fields(
+					&pcur, offsets, big_rec, &mtr,
+					BTR_STORE_UPDATE);
+			}
+
+			dtuple_big_rec_free(big_rec);
+		}
+		if (offsets_heap) {
+			mem_heap_free(offsets_heap);
+		}
+		btr_pcur_close(&pcur);
+	}
+
+func_exit:
+	mutex_exit(&dict_sys.mutex);
+	mtr.commit();
+	mem_heap_free(heap);
+
+	return err != DB_SUCCESS;
+}
+
 /** Adjust the create index column number from "New table" to
 "old InnoDB table" while we are doing dropping virtual column. Since we do
 not create separate new table for the dropping/adding virtual columns.
@@ -6569,7 +6877,7 @@ new_clustered_failed:
 	DBUG_ASSERT(!ctx->need_rebuild()
 		    || !ctx->new_table->persistent_autoinc);
 
-	if (ctx->need_rebuild() && instant_alter_column_possible(
+	if (ctx->need_rebuild() && instant_alter_possible(
 		    *user_table, ha_alter_info, old_table, altered_table,
 		    ha_innobase::is_innodb_strict_mode(ctx->trx->mysql_thd))) {
 		for (uint a = 0; a < ctx->num_to_add_index; a++) {
@@ -8065,7 +8373,8 @@ err_exit:
 					(ha_alter_info->ignore
 					 || !thd_is_strict_mode(m_user_thd)),
 					alt_opt.page_compressed,
-					alt_opt.page_compression_level);
+					alt_opt.page_compression_level,
+					alt_opt.persistent_count);
 		}
 
 		DBUG_ASSERT(m_prebuilt->trx->dict_operation_lock_mode == 0);
@@ -8190,7 +8499,8 @@ found_col:
 		ha_alter_info->create_info->auto_increment_value,
 		autoinc_col_max_value,
 		ha_alter_info->ignore || !thd_is_strict_mode(m_user_thd),
-		alt_opt.page_compressed, alt_opt.page_compression_level);
+		alt_opt.page_compressed, alt_opt.page_compression_level,
+		alt_opt.persistent_count);
 
 	DBUG_RETURN(prepare_inplace_alter_table_dict(
 			    ha_alter_info, altered_table, table,
@@ -8339,6 +8649,12 @@ ok_exit:
 	DBUG_ASSERT(ctx->trx);
 	DBUG_ASSERT(ctx->prebuilt == m_prebuilt);
 
+	if (ctx->persistent_count) {
+		enable_persistent_count();
+	} else {
+		disable_persistent_count();
+	}
+
 	if (ctx->is_instant()) goto ok_exit;
 
 	dict_index_t*	pk = dict_table_get_first_index(m_prebuilt->table);
diff --git a/storage/innobase/include/btr0btr.h b/storage/innobase/include/btr0btr.h
index 7fae1ad163b..727b400bca3 100644
--- a/storage/innobase/include/btr0btr.h
+++ b/storage/innobase/include/btr0btr.h
@@ -176,7 +176,6 @@ record is in spatial index */
 /** Report that an index page is corrupted.
 @param[in]	buffer block
 @param[in]	index tree */
-ATTRIBUTE_COLD ATTRIBUTE_NORETURN __attribute__((nonnull))
 void btr_corruption_report(const buf_block_t* block,const dict_index_t* index);
 
 /** Assert that a B-tree page is not corrupted.
diff --git a/storage/innobase/include/dict0dict.h b/storage/innobase/include/dict0dict.h
index f7c4d5dca29..92572f2f658 100644
--- a/storage/innobase/include/dict0dict.h
+++ b/storage/innobase/include/dict0dict.h
@@ -867,7 +867,8 @@ inline size_t dict_table_t::get_overflow_field_local_len() const
 @param[in]	zip_ssize	Zip Shift Size
 @param[in]	use_data_dir	Table uses DATA DIRECTORY
 @param[in]	page_compressed Table uses page compression
-@param[in]	page_compression_level Page compression level */
+@param[in]	page_compression_level Page compression level
+@param[in]	persistent_count Table uses persistent count */
 UNIV_INLINE
 void
 dict_tf_set(
@@ -876,7 +877,8 @@ dict_tf_set(
 	ulint		zip_ssize,
 	bool		use_data_dir,
 	bool		page_compressed,
-	ulint		page_compression_level);
+	ulint		page_compression_level,
+	bool		persistent_count);
 
 /** Convert a 32 bit integer table flags to the 32 bit FSP Flags.
 Fsp Flags are written into the tablespace header at the offset
diff --git a/storage/innobase/include/dict0dict.ic b/storage/innobase/include/dict0dict.ic
index eda639ba7c1..5d6968dfe12 100644
--- a/storage/innobase/include/dict0dict.ic
+++ b/storage/innobase/include/dict0dict.ic
@@ -559,7 +559,8 @@ dict_tf_get_rec_format(
 @param[in]	zip_ssize	Zip Shift Size
 @param[in]	use_data_dir	Table uses DATA DIRECTORY
 @param[in]	page_compressed Table uses page compression
-@param[in]	page_compression_level Page compression level */
+@param[in]	page_compression_level Page compression level
+@param[in]	persistent_count Table uses persistent count */
 UNIV_INLINE
 void
 dict_tf_set(
@@ -569,7 +570,8 @@ dict_tf_set(
 	ulint		zip_ssize,
 	bool		use_data_dir,
 	bool		page_compressed,
-	ulint		page_compression_level)
+	ulint		page_compression_level,
+	bool		persistent_count)
 {
 	*flags = use_data_dir ? 1 << DICT_TF_POS_DATA_DIR : 0;
 
@@ -604,6 +606,10 @@ dict_tf_set(
 		ut_ad(dict_tf_get_page_compression(*flags) == TRUE);
 		ut_ad(dict_tf_get_page_compression_level(*flags) == page_compression_level);
 	}
+
+	if (persistent_count) {
+		*flags |= (1 << DICT_TF_POS_PERSISTENT_COUNT);
+	}
 }
 
 /** Convert a 32 bit integer table flags to the 32 bit FSP Flags.
diff --git a/storage/innobase/include/dict0mem.h b/storage/innobase/include/dict0mem.h
index 1ad2517c8fb..bd9e589e6c2 100644
--- a/storage/innobase/include/dict0mem.h
+++ b/storage/innobase/include/dict0mem.h
@@ -146,6 +146,11 @@ ATOMIC_WRITES=ON and ATOMIC_WRITES=OFF between MariaDB 10.1.0 and 10.2.3)
 */
 #define DICT_TF_WIDTH_NO_ROLLBACK 2
 
+/**
+Width of the persistent count flag
+*/
+#define DICT_TF_WIDTH_PERSISTENT_COUNT 1
+
 /** Width of all the currently known table flags */
 #define DICT_TF_BITS	(DICT_TF_WIDTH_COMPACT			\
 			+ DICT_TF_WIDTH_ZIP_SSIZE		\
@@ -153,7 +158,8 @@ ATOMIC_WRITES=ON and ATOMIC_WRITES=OFF between MariaDB 10.1.0 and 10.2.3)
 			+ DICT_TF_WIDTH_DATA_DIR		\
 			+ DICT_TF_WIDTH_PAGE_COMPRESSION	\
 			+ DICT_TF_WIDTH_PAGE_COMPRESSION_LEVEL	\
-			+ DICT_TF_WIDTH_NO_ROLLBACK)
+			+ DICT_TF_WIDTH_NO_ROLLBACK	  \
+			+ DICT_TF_WIDTH_PERSISTENT_COUNT)
 
 /** Zero relative shift position of the COMPACT field */
 #define DICT_TF_POS_COMPACT		0
@@ -175,8 +181,11 @@ ATOMIC_WRITES=ON and ATOMIC_WRITES=OFF between MariaDB 10.1.0 and 10.2.3)
 /** Zero relative shift position of the NO_ROLLBACK field */
 #define DICT_TF_POS_NO_ROLLBACK		(DICT_TF_POS_PAGE_COMPRESSION_LEVEL \
 					+ DICT_TF_WIDTH_PAGE_COMPRESSION_LEVEL)
-#define DICT_TF_POS_UNUSED		(DICT_TF_POS_NO_ROLLBACK     \
+/** Zero relative shift position of the PERSISTENT_COUNT field */
+#define DICT_TF_POS_PERSISTENT_COUNT (DICT_TF_POS_NO_ROLLBACK     \
 					+ DICT_TF_WIDTH_NO_ROLLBACK)
+#define DICT_TF_POS_UNUSED           (DICT_TF_POS_PERSISTENT_COUNT    \
+					+ DICT_TF_WIDTH_PERSISTENT_COUNT)
 
 /** Bit mask of the COMPACT field */
 #define DICT_TF_MASK_COMPACT				\
@@ -206,6 +215,11 @@ ATOMIC_WRITES=ON and ATOMIC_WRITES=OFF between MariaDB 10.1.0 and 10.2.3)
 #define DICT_TF_MASK_NO_ROLLBACK		\
 		((~(~0U << DICT_TF_WIDTH_NO_ROLLBACK)) \
 		<< DICT_TF_POS_NO_ROLLBACK)
+/** Bit mask of the PERSISTENT_COUNT field */
+#define DICT_TF_MASK_PERSISTENT_COUNT		\
+		((~(~0U << DICT_TF_WIDTH_PERSISTENT_COUNT)) \
+		<< DICT_TF_POS_PERSISTENT_COUNT)
+
 
 /** Return the value of the COMPACT field */
 #define DICT_TF_GET_COMPACT(flags)			\
@@ -231,7 +245,10 @@ ATOMIC_WRITES=ON and ATOMIC_WRITES=OFF between MariaDB 10.1.0 and 10.2.3)
 #define DICT_TF_GET_PAGE_COMPRESSION_LEVEL(flags)       \
 		((flags & DICT_TF_MASK_PAGE_COMPRESSION_LEVEL)	\
 		>> DICT_TF_POS_PAGE_COMPRESSION_LEVEL)
-
+/** Return the value of the PERSISTENT_COUNT field */
+#define DICT_TF_GET_PERSISTENT_COUNT(flags)       \
+		((flags & DICT_TF_MASK_PERSISTENT_COUNT)	\
+		>> DICT_TF_POS_PERSISTENT_COUNT)
 /* @} */
 
 /** @brief Table Flags set number 2.
@@ -1852,16 +1869,23 @@ struct dict_table_t {
 		return NULL;
 	}
 
-	/** Serialise metadata of dropped or reordered columns.
+	/** Metadata BLOB structure. */
+	#define NUM_NON_PK_FIELDS_SIZE 4
+	#define NON_PK_FIELD_SIZE 2
+	#define COMMITTED_COUNT_SIZE 8
+
+	/** Serialise metadata BLOB, consisting of dropped or reordered columns,
+	and committed count.
 	@param[in,out]	heap	memory heap for allocation
 	@param[out]	field	data field with the metadata */
-	inline void serialise_columns(mem_heap_t* heap, dfield_t* field) const;
+	inline void serialise_mblob(mem_heap_t* heap, dfield_t* field) const;
 
-	/** Reconstruct dropped or reordered columns.
-	@param[in]	metadata	data from serialise_columns()
+	/** Deserialise metadata BLOB and reconstruct dropped or reordered columns,
+	and committed count.
+	@param[in]	metadata	data from serialise_mblob()
 	@param[in]	len		length of the metadata, in bytes
 	@return whether parsing the metadata failed */
-	bool deserialise_columns(const byte* metadata, ulint len);
+	bool deserialise_mblob(const byte* metadata, ulint len);
 
 	/** Set is_instant() before instant_column().
 	@param[in]	old		previous table definition
@@ -1871,7 +1895,8 @@ struct dict_table_t {
 					1 + first changed column position */
 	inline void prepare_instant(const dict_table_t& old,
 				    const ulint* col_map,
-				    unsigned& first_alter_pos);
+				    unsigned& first_alter_pos,
+				    bool alter_persistent_count);
 
 	/** Adjust table metadata for instant ADD/DROP/reorder COLUMN.
 	@param[in]	table	table on which prepare_instant() was invoked
@@ -2293,6 +2318,12 @@ public:
 	It is protected by lock_sys.mutex. */
 	ulint					n_rec_locks;
 
+	/* Whether committed count is initialized. */
+	std::atomic<bool>       committed_count_inited;
+
+	/* Count of committed records. */
+	Atomic_counter<uint64_t>       committed_count;
+
 private:
 	/** Count of how many handles are opened to this table. Dropping of the
 	table is NOT allowed until this count gets to zero. MySQL does NOT
diff --git a/storage/innobase/include/trx0trx.h b/storage/innobase/include/trx0trx.h
index 2aa0ed1e0e3..4e71d5edc11 100644
--- a/storage/innobase/include/trx0trx.h
+++ b/storage/innobase/include/trx0trx.h
@@ -705,6 +705,12 @@ struct trx_rsegs_t {
 	trx_temp_undo_t	m_noredo;
 };
 
+/** Uncommitted difference for an undo log rec */
+struct undo_rec_diff_t {
+	int64_t diff;
+	table_id_t table_id;
+};
+
 struct trx_t : ilist_node<> {
 private:
   /**
@@ -1049,6 +1055,9 @@ private:
 public:
   /** Commit the transaction. */
   void commit();
+  /** Return number of uncommitted records for table within transaction
+  @param[in]	table 	table to count uncommitted records for */
+  int64_t uncommitted_count(const dict_table_t* table) const;
 
 
   bool is_referenced() const { return n_ref > 0; }
@@ -1168,7 +1177,6 @@ struct commit_node_t{
 			state;	/*!< node execution state */
 };
 
-
 /** Test if trx->mutex is owned. */
 #define trx_mutex_own(t) mutex_own(&t->mutex)
 
diff --git a/storage/innobase/rem/rem0rec.cc b/storage/innobase/rem/rem0rec.cc
index e98448624b3..976008132b0 100644
--- a/storage/innobase/rem/rem0rec.cc
+++ b/storage/innobase/rem/rem0rec.cc
@@ -604,7 +604,7 @@ rec_init_offsets(
 	call in btr_cur_instant_init_low(). We cannot invoke
 	index->is_instant(), because the same assertion would fail there
 	until btr_cur_instant_init_low() has invoked
-	dict_table_t::deserialise_columns(). */
+	dict_table_t::deserialise_mblob(). */
 	ut_ad(index->n_core_null_bytes <= UT_BITS_IN_BYTES(index->n_nullable)
 	      || index->in_instant_init);
 	ut_d(memcpy(&offsets[RECORD_OFFSET], &rec, sizeof(rec)));
diff --git a/storage/innobase/trx/trx0trx.cc b/storage/innobase/trx/trx0trx.cc
index bd54af76d9d..443f9fd1db8 100644
--- a/storage/innobase/trx/trx0trx.cc
+++ b/storage/innobase/trx/trx0trx.cc
@@ -47,6 +47,7 @@ Created 3/26/1996 Heikki Tuuri
 #include "trx0xa.h"
 #include "ut0pool.h"
 #include "ut0vec.h"
+#include "ha_innodb.h"
 
 #include <set>
 #include <new>
@@ -1492,6 +1493,108 @@ inline void trx_t::commit_in_memory(const mtr_t *mtr)
     srv_wake_purge_thread_if_not_active();
 }
 
+/** Get difference in uncommitted count for a single undo record.
+@param[in] undo_rec    undo record
+@param[out] undo_rec_diff    undo record difference */
+void undo_rec_get_diff(trx_undo_rec_t* undo_rec,
+                       undo_rec_diff_t* undo_rec_diff)
+{
+	ulint type, cmpl_info;
+	bool updated_extern;
+	undo_no_t undo;
+
+	trx_undo_rec_get_pars(undo_rec, &type, &cmpl_info, &updated_extern, &undo,
+		&undo_rec_diff->table_id);
+
+	switch (type) {
+	case TRX_UNDO_INSERT_REC:
+	case TRX_UNDO_UPD_DEL_REC:
+		undo_rec_diff->diff = 1;
+		break;
+	case TRX_UNDO_DEL_MARK_REC:
+		undo_rec_diff->diff = -1;
+		break;
+	default:
+		undo_rec_diff->diff = 0;
+		break;
+	}
+}
+
+/** Update the persistent counts of all tables modified by a transaction.
+@param[in,out] trx    transaction */
+static void trx_update_persistent_counts(trx_t* trx)
+{
+    /* Initialize uncommitted_counts and mod_tables_by_id maps.
+    uncommitted_counts maps table ID to uncommitted count.
+    mod_tables_by_id maps table ID to dict_table_t. */
+    dict_table_t* ib_table;
+    std::map<table_id_t, int64_t> uncommitted_counts;
+    std::map<table_id_t, dict_table_t* const> mod_tables_by_id;
+    for (trx_mod_tables_t::const_iterator it = trx->mod_tables.begin();
+         it != trx->mod_tables.end(); it++) {
+        ib_table = it->first;
+        if (ib_table->committed_count_inited) {
+            uncommitted_counts.insert(
+                std::pair<table_id_t, int64_t>
+                    (ib_table->id, 0));
+            mod_tables_by_id.insert(
+                std::pair<table_id_t, dict_table_t* const>
+                    (ib_table->id, ib_table));
+        }
+    }
+
+    /* Loop through undo log and update uncommitted_counts map */
+    trx_undo_t* undo;
+    trx_undo_rec_t* undo_rec;
+    mtr_t mtr;
+    undo_rec_diff_t undo_rec_diff;
+    buf_block_t* undo_page;
+    std::map<table_id_t, int64_t>::iterator it_uncommitted_counts;
+
+    undo = trx->rsegs.m_redo.undo;
+    if (undo && !uncommitted_counts.empty()) {
+        mtr_start(&mtr);
+
+        undo_rec = trx_undo_get_first_rec(*undo->rseg->space, undo->hdr_page_no,
+                    undo->hdr_offset, RW_S_LATCH, undo_page, &mtr);
+
+        while (undo_rec) {
+            undo_rec_get_diff(undo_rec, &undo_rec_diff);
+            it_uncommitted_counts =
+                uncommitted_counts.find(undo_rec_diff.table_id);
+            if (it_uncommitted_counts != uncommitted_counts.end()) {
+                it_uncommitted_counts->second += undo_rec_diff.diff;
+            }
+
+            undo_rec = trx_undo_get_next_rec(undo_page, page_offset(undo_rec),
+				undo->hdr_page_no, undo->hdr_offset, &mtr);
+        }
+        mtr_commit(&mtr);
+    }
+
+    /* Loop through uncommitted_counts map and update corresponding ib_table
+    in mod_tables_by_id */
+    std::map<table_id_t, dict_table_t* const>::iterator it_mod_tables_by_id;
+    for (it_uncommitted_counts = uncommitted_counts.begin();
+         it_uncommitted_counts != uncommitted_counts.end();
+         it_uncommitted_counts++) {
+        it_mod_tables_by_id = mod_tables_by_id.find(
+                                it_uncommitted_counts->first);
+        ut_ad(it_mod_tables_by_id != mod_tables_by_id.end());
+        ib_table = it_mod_tables_by_id->second;
+
+        /* No guarantee that persistent count wasn't disabled between this
+        function's initial check and now */
+        if (ib_table->committed_count_inited) {
+            ib_table->committed_count += it_uncommitted_counts->second;
+            if (it_uncommitted_counts->second != 0) {
+                innobase_update_persistent_count(ib_table, trx);
+            }
+        }
+    }
+}
+
+
 /** Commit the transaction in a mini-transaction.
 @param mtr  mini-transaction (if there are any persistent modifications) */
 void trx_t::commit_low(mtr_t *mtr)
@@ -1519,6 +1622,8 @@ void trx_t::commit_low(mtr_t *mtr)
   const bool debug_sync= mysql_thd && has_logged_persistent();
 #endif
 
+  trx_update_persistent_counts(this);
+
   if (mtr)
   {
     trx_write_serialisation_history(this, mtr);
@@ -2352,3 +2457,38 @@ trx_set_rw_mode(
 		trx->read_view.set_creator_trx_id(trx->id);
 	}
 }
+
+/** Determine the change to uncommitted records for a table.
+@param[in] table     persistent table
+@return the change to uncommitted records for a table in the transaction */
+int64_t
+trx_t::uncommitted_count(const dict_table_t* table) const
+{
+    trx_undo_t* undo;
+    trx_undo_rec_t* undo_rec;
+    mtr_t mtr;
+    undo_rec_diff_t undo_rec_diff;
+    buf_block_t* undo_page;
+    int64_t count = 0;
+
+    undo = rsegs.m_redo.undo;
+    if (undo) {
+		mtr_start(&mtr);
+
+        undo_rec = trx_undo_get_first_rec(*undo->rseg->space, undo->hdr_page_no,
+            undo->hdr_offset, RW_S_LATCH, undo_page, &mtr);
+
+        while (undo_rec) {
+            undo_rec_get_diff(undo_rec, &undo_rec_diff);
+            if (undo_rec_diff.table_id == table->id) {
+                count += undo_rec_diff.diff;
+            }
+
+            undo_rec = trx_undo_get_next_rec(undo_page, page_offset(undo_rec),
+				undo->hdr_page_no, undo->hdr_offset, &mtr);
+        }
+        mtr_commit(&mtr);
+    }
+
+    return count;
+}