Merge 10.4 into 10.5st-10.5-merge

3 files changed, 474 insertions, 144 deletions

diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 7d3eb64e5a0..e61c19cd0d5 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -13163,25 +13163,6 @@ ha_innobase::discard_or_import_tablespace(
 				    err, m_prebuilt->table->flags, NULL));
 	}
 
-	/* Evict and reload the table definition in order to invoke
-	btr_cur_instant_init(). */
-	table_id_t id = m_prebuilt->table->id;
-	ut_ad(id);
-	mutex_enter(&dict_sys.mutex);
-	dict_table_close(m_prebuilt->table, TRUE, FALSE);
-	dict_sys.remove(m_prebuilt->table);
-	m_prebuilt->table = dict_table_open_on_id(id, TRUE,
-						  DICT_TABLE_OP_NORMAL);
-	mutex_exit(&dict_sys.mutex);
-	if (!m_prebuilt->table) {
-		err = DB_TABLE_NOT_FOUND;
-	} else {
-		if (const Field* ai = table->found_next_number_field) {
-			initialize_auto_increment(m_prebuilt->table, ai);
-		}
-		dict_stats_init(m_prebuilt->table);
-	}
-
 	if (dict_stats_is_persistent_enabled(m_prebuilt->table)) {
 		dberr_t		ret;
 
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index 10fec4d0871..ea4a0b93516 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -633,6 +633,18 @@ public:
 	static bool full_crc32(ulint flags) {
 		return flags & FSP_FLAGS_FCRC32_MASK_MARKER;
 	}
+  /** Determine if full_crc32 is used along with compression */
+  static bool is_full_crc32_compressed(ulint flags)
+  {
+    if (full_crc32(flags))
+    {
+      ulint algo= FSP_FLAGS_FCRC32_GET_COMPRESSED_ALGO(flags);
+      DBUG_ASSERT(algo <= PAGE_ALGORITHM_LAST);
+      return algo > 0;
+    }
+
+    return false;
+  }
 	/** @return whether innodb_checksum_algorithm=full_crc32 is active */
 	bool full_crc32() const { return full_crc32(flags); }
 	/** Determine the logical page size.
@@ -694,21 +706,15 @@ public:
 	unsigned zip_size() const { return zip_size(flags); }
 	/** @return the physical page size */
 	unsigned physical_size() const { return physical_size(flags); }
-	/** Check whether the compression enabled in tablespace.
-	@param[in]	flags	tablespace flags */
-	static bool is_compressed(ulint flags) {
-
-		if (full_crc32(flags)) {
-			ulint algo = FSP_FLAGS_FCRC32_GET_COMPRESSED_ALGO(
-				flags);
-			DBUG_ASSERT(algo <= PAGE_ALGORITHM_LAST);
-			return algo > 0;
-		}
-
-		return FSP_FLAGS_HAS_PAGE_COMPRESSION(flags);
-	}
-	/** @return whether the compression enabled for the tablespace. */
-	bool is_compressed() const { return is_compressed(flags); }
+  /** Check whether the compression enabled in tablespace.
+  @param[in]	flags	tablespace flags */
+  static bool is_compressed(ulint flags)
+  {
+    return is_full_crc32_compressed(flags) ||
+      FSP_FLAGS_HAS_PAGE_COMPRESSION(flags);
+  }
+  /** @return whether the compression enabled for the tablespace. */
+  bool is_compressed() const { return is_compressed(flags); }
 
 	/** Get the compression algorithm for full crc32 format.
 	@param[in]	flags	tablespace flags
diff --git a/storage/innobase/row/row0import.cc b/storage/innobase/row/row0import.cc
index 7fe5cc1802c..aa1613fe436 100644
--- a/storage/innobase/row/row0import.cc
+++ b/storage/innobase/row/row0import.cc
@@ -40,6 +40,7 @@ Created 2012-02-08 by Sunny Bains.
 #include "row0quiesce.h"
 #include "fil0pagecompress.h"
 #include "trx0undo.h"
+#include "row0row.h"
 #ifdef HAVE_LZO
 #include "lzo/lzo1x.h"
 #endif
@@ -47,6 +48,8 @@ Created 2012-02-08 by Sunny Bains.
 #include "snappy-c.h"
 #endif
 
+#include "scope.h"
+
 #include <vector>
 
 #ifdef HAVE_MY_AES_H
@@ -190,6 +193,9 @@ struct row_import {
 	dberr_t match_schema(
 		THD*			thd) UNIV_NOTHROW;
 
+	dberr_t match_flags(THD *thd) const ;
+
+
 	dict_table_t*	m_table;		/*!< Table instance */
 
 	ulint		m_version;		/*!< Version of config file */
@@ -327,7 +333,7 @@ public:
 	mtr_t		m_mtr;
 };
 
-/** Class that purges delete marked reocords from indexes, both secondary
+/** Class that purges delete marked records from indexes, both secondary
 and cluster. It does a pessimistic delete. This should only be done if we
 couldn't purge the delete marked reocrds during Phase I. */
 class IndexPurge {
@@ -605,8 +611,6 @@ AbstractCallback::init(
 	if (!fil_space_t::is_valid_flags(m_space_flags, true)) {
 		ulint cflags = fsp_flags_convert_from_101(m_space_flags);
 		if (cflags == ULINT_UNDEFINED) {
-			ib::error() << "Invalid FSP_SPACE_FLAGS="
-				<< ib::hex(m_space_flags);
 			return(DB_CORRUPTION);
 		}
 		m_space_flags = cflags;
@@ -1123,7 +1127,8 @@ row_import::match_index_columns(
 
 	for (ulint i = 0; i < index->n_fields; ++i, ++field, ++cfg_field) {
 
-		if (strcmp(field->name(), cfg_field->name()) != 0) {
+		if (field->name() && cfg_field->name()
+		     && strcmp(field->name(), cfg_field->name()) != 0) {
 			ib_errf(thd, IB_LOG_LEVEL_ERROR,
 				ER_TABLE_SCHEMA_MISMATCH,
 				"Index field name %s doesn't match"
@@ -1292,6 +1297,61 @@ row_import::match_table_columns(
 	return(err);
 }
 
+dberr_t row_import::match_flags(THD *thd) const
+{
+  ulint mismatch= (m_table->flags ^ m_flags) & ~DICT_TF_MASK_DATA_DIR;
+  if (!mismatch)
+    return DB_SUCCESS;
+
+  const char *msg;
+  if (mismatch & DICT_TF_MASK_ZIP_SSIZE)
+  {
+    if ((m_table->flags & DICT_TF_MASK_ZIP_SSIZE) &&
+        (m_flags & DICT_TF_MASK_ZIP_SSIZE))
+    {
+      switch (m_flags & DICT_TF_MASK_ZIP_SSIZE) {
+      case 0U << DICT_TF_POS_ZIP_SSIZE:
+        goto uncompressed;
+      case 1U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=1";
+        break;
+      case 2U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=2";
+        break;
+      case 3U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=4";
+        break;
+      case 4U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=8";
+        break;
+      case 5U << DICT_TF_POS_ZIP_SSIZE:
+        msg= "ROW_FORMAT=COMPRESSED KEY_BLOCK_SIZE=16";
+        break;
+      default:
+        msg= "strange KEY_BLOCK_SIZE";
+      }
+    }
+    else if (m_flags & DICT_TF_MASK_ZIP_SSIZE)
+      msg= "ROW_FORMAT=COMPRESSED";
+    else
+      goto uncompressed;
+  }
+  else
+  {
+  uncompressed:
+    msg= (m_flags & DICT_TF_MASK_ATOMIC_BLOBS) ? "ROW_FORMAT=DYNAMIC"
+         : (m_flags & DICT_TF_MASK_COMPACT)    ? "ROW_FORMAT=COMPACT"
+                                               : "ROW_FORMAT=REDUNDANT";
+  }
+
+  ib_errf(thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
+          "Table flags don't match, server table has 0x%x and the meta-data "
+          "file has 0x%zx; .cfg file uses %s",
+          m_table->flags, m_flags, msg);
+
+  return DB_ERROR;
+}
+
 /** Check if the table (and index) schema that was read from the .cfg file
 matches the in memory table definition.
 @param thd MySQL session variable
@@ -1302,60 +1362,7 @@ row_import::match_schema(
 {
 	/* Do some simple checks. */
 
-	if (ulint mismatch = (m_table->flags ^ m_flags)
-	    & ~DICT_TF_MASK_DATA_DIR) {
-		const char* msg;
-		if (mismatch & DICT_TF_MASK_ZIP_SSIZE) {
-			if ((m_table->flags & DICT_TF_MASK_ZIP_SSIZE)
-			    && (m_flags & DICT_TF_MASK_ZIP_SSIZE)) {
-				switch (m_flags & DICT_TF_MASK_ZIP_SSIZE) {
-				case 0U << DICT_TF_POS_ZIP_SSIZE:
-					goto uncompressed;
-				case 1U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=1";
-					break;
-				case 2U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=2";
-					break;
-				case 3U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=4";
-					break;
-				case 4U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=8";
-					break;
-				case 5U << DICT_TF_POS_ZIP_SSIZE:
-					msg = "ROW_FORMAT=COMPRESSED"
-						" KEY_BLOCK_SIZE=16";
-					break;
-				default:
-					msg = "strange KEY_BLOCK_SIZE";
-				}
-			} else if (m_flags & DICT_TF_MASK_ZIP_SSIZE) {
-				msg = "ROW_FORMAT=COMPRESSED";
-			} else {
-				goto uncompressed;
-			}
-		} else {
-uncompressed:
-			msg = (m_flags & DICT_TF_MASK_ATOMIC_BLOBS)
-				? "ROW_FORMAT=DYNAMIC"
-				: (m_flags & DICT_TF_MASK_COMPACT)
-				? "ROW_FORMAT=COMPACT"
-				: "ROW_FORMAT=REDUNDANT";
-		}
-
-		ib_errf(thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
-			"Table flags don't match, server table has 0x%x"
-			" and the meta-data file has 0x" ULINTPFx ";"
-			" .cfg file uses %s",
-			m_table->flags, m_flags, msg);
-
-		return(DB_ERROR);
-	} else if (m_table->n_cols != m_n_cols) {
+	if (m_table->n_cols != m_n_cols) {
 		ib_errf(thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
 			"Number of columns don't match, table has %u "
 			"columns but the tablespace meta-data file has "
@@ -1977,42 +1984,7 @@ PageConverter::update_index_page(
 	}
 
 	if (m_index->m_srv_index->is_clust()) {
-		if (page_id.page_no() == m_index->m_srv_index->page) {
-			dict_index_t* index = const_cast<dict_index_t*>(
-				m_index->m_srv_index);
-			/* Preserve the PAGE_ROOT_AUTO_INC. */
-			if (index->table->supports_instant()) {
-				if (btr_cur_instant_root_init(index, page)) {
-					return(DB_CORRUPTION);
-				}
-
-				if (index->n_core_fields > index->n_fields) {
-					/* Some columns have been dropped.
-					Refuse to IMPORT TABLESPACE for now.
-
-					NOTE: This is not an accurate check.
-					Columns could have been both
-					added and dropped instantly.
-					For an accurate check, we must read
-					the metadata BLOB page pointed to
-					by the leftmost leaf page.
-
-					But we would have to read
-					those pages in a special way,
-					bypassing the buffer pool! */
-					return DB_UNSUPPORTED;
-				}
-
-				/* Provisionally set all instantly
-				added columns to be DEFAULT NULL. */
-				for (unsigned i = index->n_core_fields;
-				     i < index->n_fields; i++) {
-					dict_col_t* col = index->fields[i].col;
-					col->def_val.len = UNIV_SQL_NULL;
-					col->def_val.data = NULL;
-				}
-			}
-		} else {
+		if (page_id.page_no() != m_index->m_srv_index->page) {
 			goto clear_page_max_trx_id;
 		}
 	} else if (page_is_leaf(page)) {
@@ -3120,6 +3092,377 @@ row_import_read_meta_data(
 	return(DB_ERROR);
 }
 
+#define BTR_BLOB_HDR_PART_LEN 0 /*!< BLOB part len on this page */
+#define BTR_BLOB_HDR_NEXT_PAGE_NO 4 /*!< next BLOB part page no,
+                                    FIL_NULL if none */
+#define BTR_BLOB_HDR_SIZE 8 /*!< Size of a BLOB part header, in bytes */
+
+/* decrypt and decompress page if needed */
+static dberr_t decrypt_decompress(fil_space_crypt_t *space_crypt,
+                                  size_t space_flags, span<byte> page,
+                                  size_t space_id, byte *page_compress_buf)
+{
+  auto *data= page.data();
+
+  if (space_crypt && space_crypt->should_encrypt())
+  {
+    if (!buf_page_verify_crypt_checksum(data, space_flags))
+      return DB_CORRUPTION;
+
+    dberr_t err;
+    if (!fil_space_decrypt(space_id, space_crypt, data, page.size(),
+                           space_flags, data, &err) || err != DB_SUCCESS)
+      return err;
+  }
+
+  bool page_compressed= false;
+
+  if (fil_space_t::full_crc32(space_flags) &&
+      fil_space_t::is_compressed(space_flags))
+    page_compressed= buf_page_is_compressed(data, space_flags);
+  else
+  {
+    switch (fil_page_get_type(data)) {
+    case FIL_PAGE_PAGE_COMPRESSED:
+    case FIL_PAGE_PAGE_COMPRESSED_ENCRYPTED:
+      page_compressed= true;
+    }
+  }
+
+  if (page_compressed)
+  {
+    auto compress_length=
+      fil_page_decompress(page_compress_buf, data, space_flags);
+    ut_ad(compress_length != srv_page_size);
+
+    if (compress_length == 0)
+      return DB_CORRUPTION;
+  }
+
+  return DB_SUCCESS;
+}
+
+static size_t get_buf_size()
+{
+  return srv_page_size
+#ifdef HAVE_LZO
+         + LZO1X_1_15_MEM_COMPRESS
+#elif defined HAVE_SNAPPY
+         + snappy_max_compressed_length(srv_page_size)
+#endif
+      ;
+}
+
+/* find, parse instant metadata, performing variaous checks,
+and apply it to dict_table_t
+@return DB_SUCCESS or some error */
+static dberr_t handle_instant_metadata(dict_table_t *table,
+                                       const row_import &cfg)
+{
+  dict_get_and_save_data_dir_path(table, false);
+
+  char *filepath;
+  if (DICT_TF_HAS_DATA_DIR(table->flags))
+  {
+    ut_a(table->data_dir_path);
+
+    filepath=
+        fil_make_filepath(table->data_dir_path, table->name.m_name, IBD, true);
+  }
+  else
+    filepath= fil_make_filepath(nullptr, table->name.m_name, IBD, false);
+
+  if (!filepath)
+    return DB_OUT_OF_MEMORY;
+
+  SCOPE_EXIT([filepath]() { ut_free(filepath); });
+
+  bool success;
+  auto file= os_file_create_simple_no_error_handling(
+      innodb_data_file_key, filepath, OS_FILE_OPEN, OS_FILE_READ_WRITE, false,
+      &success);
+  if (!success)
+    return DB_IO_ERROR;
+
+  if (os_file_get_size(file) < srv_page_size * 4)
+    return DB_CORRUPTION;
+
+  SCOPE_EXIT([&file]() { os_file_close(file); });
+
+  std::unique_ptr<byte[], decltype(&aligned_free)> first_page(
+      static_cast<byte *>(aligned_malloc(srv_page_size, srv_page_size)),
+      &aligned_free);
+
+  if (dberr_t err= os_file_read_no_error_handling(IORequestReadPartial,
+                                                  file, first_page.get(), 0,
+                                                  srv_page_size, nullptr))
+    return err;
+
+  auto space_flags= fsp_header_get_flags(first_page.get());
+
+  if (!fil_space_t::is_valid_flags(space_flags, true))
+  {
+    auto cflags= fsp_flags_convert_from_101(space_flags);
+    if (cflags == ULINT_UNDEFINED)
+    {
+      ib::error() << "Invalid FSP_SPACE_FLAGS=" << ib::hex(space_flags);
+      return DB_CORRUPTION;
+    }
+    space_flags= static_cast<decltype(space_flags)>(cflags);
+  }
+
+  if (!cfg.m_missing)
+  {
+    if (dberr_t err= cfg.match_flags(current_thd))
+      return err;
+  }
+
+  const auto zip_size= fil_space_t::zip_size(space_flags);
+  const uint64_t physical_size= zip_size ? zip_size : srv_page_size;
+  ut_ad(physical_size <= UNIV_PAGE_SIZE_MAX);
+  const auto space_id= page_get_space_id(first_page.get());
+
+  auto *space_crypt= fil_space_read_crypt_data(zip_size, first_page.get());
+  SCOPE_EXIT([&space_crypt]() {
+    if (space_crypt)
+      fil_space_destroy_crypt_data(&space_crypt);
+  });
+
+  std::unique_ptr<byte[], decltype(&aligned_free)> page(
+      static_cast<byte *>(
+          aligned_malloc(UNIV_PAGE_SIZE_MAX, UNIV_PAGE_SIZE_MAX)),
+      &aligned_free);
+
+  if (dberr_t err= os_file_read_no_error_handling(
+          IORequestReadPartial, file, page.get(), 3 * physical_size,
+          physical_size, nullptr))
+    return err;
+
+  std::unique_ptr<byte[]> page_compress_buf(new byte[get_buf_size()]);
+
+  if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
+                                      {page.get(), static_cast<size_t>
+                                       (physical_size)},
+                                      space_id, page_compress_buf.get()))
+    return err;
+
+  if (table->supports_instant())
+  {
+    dict_index_t *index= dict_table_get_first_index(table);
+
+    auto tmp1= table->space_id;
+    table->space_id= page_get_space_id(page.get());
+    SCOPE_EXIT([tmp1, table]() { table->space_id= tmp1; });
+
+    auto tmp2= index->page;
+    index->page= page_get_page_no(page.get());
+    SCOPE_EXIT([tmp2, index]() { index->page= tmp2; });
+
+    if (!page_is_comp(page.get()) != !dict_table_is_comp(table))
+    {
+      ib_errf(current_thd, IB_LOG_LEVEL_ERROR, ER_TABLE_SCHEMA_MISMATCH,
+              "ROW_FORMAT mismatch");
+      return DB_CORRUPTION;
+    }
+
+    if (btr_cur_instant_root_init(index, page.get()))
+      return DB_ERROR;
+
+    ut_ad(index->n_core_null_bytes != dict_index_t::NO_CORE_NULL_BYTES);
+
+    if (fil_page_get_type(page.get()) == FIL_PAGE_INDEX)
+    {
+      ut_ad(!index->is_instant());
+      return DB_SUCCESS;
+    }
+
+    mem_heap_t *heap= NULL;
+    SCOPE_EXIT([&heap]() { mem_heap_free(heap); });
+
+    while (btr_page_get_level(page.get()) != 0)
+    {
+      const rec_t *rec= page_rec_get_next(page_get_infimum_rec(page.get()));
+
+      /* Relax the assertion in rec_init_offsets(). */
+      ut_ad(!index->in_instant_init);
+      ut_d(index->in_instant_init= true);
+      rec_offs *offsets=
+          rec_get_offsets(rec, index, nullptr, 0, ULINT_UNDEFINED, &heap);
+      ut_d(index->in_instant_init= false);
+
+      uint64_t child_page_no= btr_node_ptr_get_child_page_no(rec, offsets);
+
+      if (dberr_t err=
+          os_file_read_no_error_handling(IORequestReadPartial, file,
+                                         page.get(),
+                                         child_page_no * physical_size,
+                                         physical_size, nullptr))
+        return err;
+
+      if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
+                                          {page.get(), static_cast<size_t>
+                                           (physical_size)}, space_id,
+                                          page_compress_buf.get()))
+        return err;
+    }
+
+    const auto *rec= page_rec_get_next(page_get_infimum_rec(page.get()));
+    const auto comp= dict_table_is_comp(index->table);
+    const auto info_bits= rec_get_info_bits(rec, comp);
+
+    if (page_rec_is_supremum(rec) || !(info_bits & REC_INFO_MIN_REC_FLAG))
+    {
+      ib::error() << "Table " << index->table->name
+                  << " is missing instant ALTER metadata";
+      index->table->corrupted= true;
+      return DB_CORRUPTION;
+    }
+
+    if ((info_bits & ~REC_INFO_DELETED_FLAG) != REC_INFO_MIN_REC_FLAG ||
+        (comp && rec_get_status(rec) != REC_STATUS_INSTANT))
+    {
+    incompatible:
+      ib::error() << "Table " << index->table->name
+                  << " contains unrecognizable instant ALTER metadata";
+      index->table->corrupted= true;
+      return DB_CORRUPTION;
+    }
+
+    if (info_bits & REC_INFO_DELETED_FLAG)
+    {
+      ulint trx_id_offset= index->trx_id_offset;
+      ut_ad(index->n_uniq);
+
+      if (trx_id_offset)
+      {
+      }
+      else if (index->table->not_redundant())
+      {
+
+        for (uint i= index->n_uniq; i--;)
+          trx_id_offset+= index->fields[i].fixed_len;
+      }
+      else if (rec_get_1byte_offs_flag(rec))
+      {
+        trx_id_offset= rec_1_get_field_end_info(rec, index->n_uniq - 1);
+        ut_ad(!(trx_id_offset & REC_1BYTE_SQL_NULL_MASK));
+        trx_id_offset&= ~REC_1BYTE_SQL_NULL_MASK;
+      }
+      else
+      {
+        trx_id_offset= rec_2_get_field_end_info(rec, index->n_uniq - 1);
+        ut_ad(!(trx_id_offset & REC_2BYTE_SQL_NULL_MASK));
+        trx_id_offset&= ~REC_2BYTE_SQL_NULL_MASK;
+      }
+
+      const byte *ptr=
+          rec + trx_id_offset + (DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
+
+      if (mach_read_from_4(ptr + BTR_EXTERN_LEN))
+        goto incompatible;
+
+      uint len= mach_read_from_4(ptr + BTR_EXTERN_LEN + 4);
+      if (!len || mach_read_from_4(ptr + BTR_EXTERN_OFFSET) != FIL_PAGE_DATA)
+        goto incompatible;
+
+      std::unique_ptr<byte[], decltype(&aligned_free)>
+        second_page(static_cast<byte*>(aligned_malloc(physical_size,
+                                                      physical_size)),
+                    &aligned_free);
+
+      if (dberr_t err=
+          os_file_read_no_error_handling(IORequestReadPartial, file,
+                                         second_page.get(), physical_size *
+                                         mach_read_from_4(ptr +
+                                                          BTR_EXTERN_PAGE_NO),
+                                         srv_page_size, nullptr))
+        return err;
+
+      if (dberr_t err= decrypt_decompress(space_crypt, space_flags,
+                                          {second_page.get(),
+                                           static_cast<size_t>(physical_size)},
+                                          space_id, page_compress_buf.get()))
+        return err;
+
+      if (fil_page_get_type(second_page.get()) != FIL_PAGE_TYPE_BLOB ||
+          mach_read_from_4(
+              &second_page[FIL_PAGE_DATA + BTR_BLOB_HDR_NEXT_PAGE_NO]) !=
+              FIL_NULL ||
+          mach_read_from_4(
+              &second_page[FIL_PAGE_DATA + BTR_BLOB_HDR_PART_LEN]) != len)
+        goto incompatible;
+
+      /* The unused part of the BLOB page should be zero-filled. */
+      for (const byte *
+               b= second_page.get() + (FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE) +
+                  len,
+              *const end= second_page.get() + srv_page_size - BTR_EXTERN_LEN;
+           b < end;)
+      {
+        if (*b++)
+          goto incompatible;
+      }
+
+      if (index->table->deserialise_columns(
+              &second_page[FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE], len))
+        goto incompatible;
+    }
+
+    rec_offs *offsets= rec_get_offsets(
+        rec, index, nullptr, index->n_core_fields, ULINT_UNDEFINED, &heap);
+    if (rec_offs_any_default(offsets))
+    {
+    inconsistent:
+      goto incompatible;
+    }
+
+    /* In fact, because we only ever append fields to the metadata
+    record, it is also OK to perform READ UNCOMMITTED and
+    then ignore any extra fields, provided that
+    trx_sys.is_registered(DB_TRX_ID). */
+    if (rec_offs_n_fields(offsets) >
+            ulint(index->n_fields) + !!index->table->instant &&
+        !trx_sys.is_registered(current_trx(),
+                               row_get_rec_trx_id(rec, index, offsets)))
+      goto inconsistent;
+
+    for (unsigned i= index->n_core_fields; i < index->n_fields; i++)
+    {
+      dict_col_t *col= index->fields[i].col;
+      const unsigned o= i + !!index->table->instant;
+      ulint len;
+      const byte *data= rec_get_nth_field(rec, offsets, o, &len);
+      ut_ad(!col->is_added());
+      ut_ad(!col->def_val.data);
+      col->def_val.len= len;
+      switch (len) {
+      case UNIV_SQL_NULL:
+        continue;
+      case 0:
+        col->def_val.data= field_ref_zero;
+        continue;
+      }
+      ut_ad(len != UNIV_SQL_DEFAULT);
+      if (!rec_offs_nth_extern(offsets, o))
+        col->def_val.data= mem_heap_dup(index->table->heap, data, len);
+      else if (len < BTR_EXTERN_FIELD_REF_SIZE ||
+               !memcmp(data + len - BTR_EXTERN_FIELD_REF_SIZE, field_ref_zero,
+                       BTR_EXTERN_FIELD_REF_SIZE))
+      {
+        col->def_val.len= UNIV_SQL_DEFAULT;
+        goto inconsistent;
+      }
+      else
+      {
+        col->def_val.data= btr_copy_externally_stored_field(
+            &col->def_val.len, data, srv_page_size, len, index->table->heap);
+      }
+    }
+  }
+
+  return DB_SUCCESS;
+}
+
 /**
 Read the contents of the <tablename>.cfg file.
 @return DB_SUCCESS or error code. */
@@ -3428,14 +3771,7 @@ dberr_t FetchIndexRootPages::run(const fil_iterator_t& iter,
 {
   const unsigned zip_size= fil_space_t::zip_size(m_space_flags);
   const unsigned size= zip_size ? zip_size : unsigned(srv_page_size);
-  const ulint buf_size=
-#ifdef HAVE_LZO
-    LZO1X_1_15_MEM_COMPRESS+
-#elif defined HAVE_SNAPPY
-    snappy_max_compressed_length(srv_page_size) +
-#endif
-    srv_page_size;
-  byte* page_compress_buf = static_cast<byte*>(malloc(buf_size));
+  byte* page_compress_buf= static_cast<byte*>(malloc(get_buf_size()));
   const bool full_crc32 = fil_space_t::full_crc32(m_space_flags);
   bool skip_checksum_check = false;
   ut_ad(!srv_read_only_mode);
@@ -3527,14 +3863,7 @@ static dberr_t fil_iterate(
 	const ulint		size = callback.physical_size();
 	ulint			n_bytes = iter.n_io_buffers * size;
 
-	const ulint buf_size = srv_page_size
-#ifdef HAVE_LZO
-		+ LZO1X_1_15_MEM_COMPRESS
-#elif defined HAVE_SNAPPY
-		+ snappy_max_compressed_length(srv_page_size)
-#endif
-		;
-	byte* page_compress_buf = static_cast<byte*>(malloc(buf_size));
+	byte* page_compress_buf= static_cast<byte*>(malloc(get_buf_size()));
 	ut_ad(!srv_read_only_mode);
 
 	if (!page_compress_buf) {
@@ -4094,6 +4423,11 @@ row_import_for_mysql(
 
 	if (err == DB_SUCCESS) {
 
+		if (dberr_t err = handle_instant_metadata(table, cfg)) {
+			rw_lock_s_unlock(&dict_sys.latch);
+			return row_import_error(prebuilt, trx, err);
+		}
+
 		/* We have a schema file, try and match it with our
 		data dictionary. */
 
@@ -4151,6 +4485,15 @@ row_import_for_mysql(
 
 			if (err == DB_SUCCESS) {
 				err = cfg.set_root_by_heuristic();
+
+				if (err == DB_SUCCESS) {
+					if (dberr_t err =
+					    handle_instant_metadata(table,
+								    cfg)) {
+						return row_import_error(
+							prebuilt, trx, err);
+					}
+				}
 			}
 		}
 	} else {