/*****************************************************************************

Copyright (c) 1994, 2019, Oracle and/or its affiliates. All Rights Reserved.
Copyright (c) 2012, Facebook Inc.
Copyright (c) 2015, 2023, MariaDB Corporation.

This program is free software; you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by the Free Software
Foundation; version 2 of the License.

This program is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with
this program; if not, write to the Free Software Foundation, Inc.,
51 Franklin Street, Fifth Floor, Boston, MA 02110-1335 USA

*****************************************************************************/

/**************************************************//**
@file btr/btr0cur.cc
The index tree cursor

All changes that row operations make to a B-tree or the records
there must go through this module! Undo log records are written here
of every modify or insert of a clustered index record.

   NOTE!!!
To make sure we do not run out of disk space during a pessimistic
insert or update, we have to reserve 2 x the height of the index tree
many pages in the tablespace before we start the operation, because
if leaf splitting has been started, it is difficult to undo, except
by crashing the database and doing a roll-forward.

Created 10/16/1994 Heikki Tuuri
*******************************************************/

#include "btr0cur.h"
#include "row0upd.h"
#include "mtr0log.h"
#include "page0page.h"
#include "page0zip.h"
#include "rem0rec.h"
#include "rem0cmp.h"
#include "buf0lru.h"
#include "buf0rea.h"
#include "btr0btr.h"
#include "btr0sea.h"
#include "row0log.h"
#include "row0purge.h"
#include "row0upd.h"
#include "trx0rec.h"
#include "trx0roll.h"
#include "que0que.h"
#include "row0row.h"
#include "srv0srv.h"
#include "lock0lock.h"
#include "zlib.h"
#include "srv0start.h"
#include "mysql_com.h"
#include "dict0stats.h"
#include "row0ins.h"
#ifdef WITH_WSREP
#include "mysql/service_wsrep.h"
#endif /* WITH_WSREP */
#include "log.h"

/** Modification types for the B-tree operation.
    Note that the order must be DELETE, BOTH, INSERT !!
 */
enum btr_intention_t {
 BTR_INTENTION_DELETE,
 BTR_INTENTION_BOTH,
 BTR_INTENTION_INSERT
};

/** For the index->lock scalability improvement, only possibility of clear
performance regression observed was caused by grown huge history list length.
That is because the exclusive use of index->lock also worked as reserving
free blocks and read IO bandwidth with priority. To avoid huge glowing history
list as same level with previous implementation, prioritizes pessimistic tree
operations by purge as the previous, when it seems to be growing huge.

 Experimentally, the history list length starts to affect to performance
throughput clearly from about 100000. */
#define BTR_CUR_FINE_HISTORY_LENGTH 100000


#ifdef UNIV_DEBUG
/* Flag to limit optimistic insert records */
uint btr_cur_limit_optimistic_insert_debug;
/** Number of times index lock was upgraded from SX to X */
Atomic_counter<uint64_t> btr_cur_n_index_lock_upgrades{0};
/** Number of times btr_cur_pessimistic_insert() was called */
Atomic_counter<uint64_t> btr_cur_pessimistic_insert_calls{0};
/** Number of times btr_cur_pessimistic_update() was called */
Atomic_counter<uint64_t> btr_cur_pessimistic_update_calls{0};
/** Number of times btr_cur_pessimistic_delete() was called */
Atomic_counter<uint64_t> btr_cur_pessimistic_delete_calls{0};
/** Number of times DB_UNDERFLOW was returned as optimistic update error in btr_cur_pessimistic_update() */
Atomic_counter<uint64_t> btr_cur_pessimistic_update_optim_err_underflows{0};
/** Number of times DB_OVERFLOW was returned as optimistic update error in btr_cur_pessimistic_update() */
Atomic_counter<uint64_t> btr_cur_pessimistic_update_optim_err_overflows{0};
#endif /* UNIV_DEBUG */

/** innodb_index_shrink: whether InnoDB may shrink a B-tree by merging or
reorganizing pages on a record-growing UPDATE. ON (the default) keeps the
historical behavior. When OFF, btr_cur_optimistic_update() gates its
DB_UNDERFLOW return behind actual record shrinkage (so a freshly split page is
not re-merged just because an update grew a record), and
btr_cur_pessimistic_update() skips btr_cur_insert_if_possible() on the
DB_OVERFLOW fallback for a growing record when an uncompressed page cannot
satisfy BTR_CUR_PAGE_REORGANIZE_LIMIT after a reorganize, falling through to a
page split instead (only size-growing UPDATEs pay the cost of an early page
split). */
my_bool btr_cur_index_shrink;

/** In the optimistic insert, if the insert does not fit, but this much space
can be released by page reorganize, then it is reorganized */
#define BTR_CUR_PAGE_REORGANIZE_LIMIT (srv_page_size / 32)

/** The structure of a BLOB part header */
/* @{ */
/*--------------------------------------*/
#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 */

/* @} */

/*******************************************************************//**
Marks all extern fields in a record as owned by the record. This function
should be called if the delete mark of a record is removed: a not delete
marked record always owns all its extern fields. */
static
void
btr_cur_unmark_extern_fields(
/*=========================*/
 buf_block_t* block, /*!< in/out: index page */
 rec_t*  rec, /*!< in/out: record in a clustered index */
 dict_index_t* index, /*!< in: index of the page */
 const rec_offs* offsets,/*!< in: array returned by rec_get_offsets() */
 mtr_t*  mtr); /*!< in: mtr, or NULL if not logged */
/***********************************************************//**
Frees the externally stored fields for a record, if the field is mentioned
in the update vector. */
static
void
btr_rec_free_updated_extern_fields(
/*===============================*/
 dict_index_t* index, /*!< in: index of rec; the index tree MUST be
X-latched */
 rec_t*  rec, /*!< in: record */
 buf_block_t* block, /*!< in: index page of rec */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec, index) */
 const upd_t* update, /*!< in: update vector */
 bool  rollback,/*!< in: performing rollback? */
 mtr_t*  mtr); /*!< in: mini-transaction handle which contains
an X-latch to record page and to the tree */
/***********************************************************//**
Frees the externally stored fields for a record. */
static
void
btr_rec_free_externally_stored_fields(
/*==================================*/
 dict_index_t* index, /*!< in: index of the data, the index
tree MUST be X-latched */
 rec_t*  rec, /*!< in: record */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec, index) */
 buf_block_t* block, /*!< in: index page of rec */
 bool  rollback,/*!< in: performing rollback? */
 mtr_t*  mtr); /*!< in: mini-transaction handle which contains
    an X-latch to record page and to the index
tree */

/*==================== B-TREE SEARCH =========================*/

/** Load the instant ALTER TABLE metadata from the clustered index
when loading a table definition.
@param[in,out] index clustered index definition
@param[in,out] mtr mini-transaction
@return error code
@retval DB_SUCCESS if no error occurred
@retval DB_CORRUPTION if any corruption was noticed */
static dberr_t btr_cur_instant_init_low(dict_index_t* index, mtr_t* mtr)
{
 ut_ad(index->is_primary());
 ut_ad(index->table->is_readable());

 if (!index->table->supports_instant()) {
  return DB_SUCCESS;
 }

 ut_ad(index->n_core_null_bytes == dict_index_t::NO_CORE_NULL_BYTES);

 dberr_t err;
 const fil_space_t* space = index->table->space;
 if (!space) {
corrupted:
  err = DB_CORRUPTION;
unreadable:
  ib::error() << "Table " << index->table->name
       << " has an unreadable root page";
  index->table->corrupted = true;
  index->table->file_unreadable = true;
  return err;
 }

 buf_block_t* root = btr_root_block_get(index, RW_SX_LATCH, mtr, &err);
 if (!root) {
  goto unreadable;
 }

 if (btr_cur_instant_root_init(index, root->page.frame)) {
  goto corrupted;
 }

 ut_ad(index->n_core_null_bytes != dict_index_t::NO_CORE_NULL_BYTES);

 if (fil_page_get_type(root->page.frame) == FIL_PAGE_INDEX) {
  ut_ad(!index->is_instant());
  return DB_SUCCESS;
 }

 btr_cur_t cur;
 /* Relax the assertion in rec_init_offsets(). */
 ut_ad(!index->in_instant_init);
 ut_d(index->in_instant_init = true);
 err = cur.open_leaf(true, index, BTR_SEARCH_LEAF, mtr);
 ut_d(index->in_instant_init = false);
 if (err != DB_SUCCESS) {
  index->table->file_unreadable = true;
  index->table->corrupted = true;
  return err;
 }

 ut_ad(page_cur_is_before_first(&cur.page_cur));
 ut_ad(page_is_leaf(btr_cur_get_page(&cur)));

 const rec_t* rec = page_cur_move_to_next(&cur.page_cur);
 const ulint comp = dict_table_is_comp(index->table);
 const ulint info_bits = rec ? rec_get_info_bits(rec, comp) : 0;

 if (page_rec_is_supremum(rec)
     || !(info_bits & REC_INFO_MIN_REC_FLAG)) {
  if (rec && !index->is_instant()) {
   /* The FIL_PAGE_TYPE_INSTANT and PAGE_INSTANT may be
   assigned even if instant ADD COLUMN was not
   committed. Changes to these page header fields are not
   undo-logged, but changes to the hidden metadata record
   are. If the server is killed and restarted, the page
   header fields could remain set even though no metadata
record is present. */
   return DB_SUCCESS;
  }

  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;
 }

 /* Read the metadata. We can get here on server restart
 or when the table was evicted from the data dictionary cache
 and is now being accessed again.

 Here, READ COMMITTED and REPEATABLE READ should be equivalent.
 Committing the ADD COLUMN operation would acquire
 MDL_EXCLUSIVE and LOCK_X|LOCK_TABLE, which would prevent any
 concurrent operations on the table, including table eviction
from the cache. */

 if (info_bits & REC_INFO_DELETED_FLAG) {
  /* This metadata record includes a BLOB that identifies
any dropped or reordered columns. */
  ulint trx_id_offset = index->trx_id_offset;
  /* If !index->trx_id_offset, the PRIMARY KEY contains
  variable-length columns. For the metadata record,
  variable-length columns should be written with zero
  length. However, before MDEV-21088 was fixed, for
  variable-length encoded PRIMARY KEY column of type
  CHAR, we wrote more than zero bytes. That is why we
  must determine the actual length of each PRIMARY KEY
  column.  The DB_TRX_ID will start right after any
PRIMARY KEY columns. */
  ut_ad(index->n_uniq);

  /* We cannot invoke rec_get_offsets() before
  index->table->deserialise_columns(). Therefore,
we must duplicate some logic here. */
  if (trx_id_offset) {
  } else if (index->table->not_redundant()) {
   /* The PRIMARY KEY contains variable-length columns.
   For the metadata record, variable-length columns are
   always written with zero length. The DB_TRX_ID will
start right after any fixed-length columns. */

   /* OK, before MDEV-21088 was fixed, for
   variable-length encoded PRIMARY KEY column of
   type CHAR, we wrote more than zero bytes. In
   order to allow affected tables to be accessed,
   it would be nice to determine the actual
   length of each PRIMARY KEY column. However, to
   be able to do that, we should determine the
   size of the null-bit bitmap in the metadata
   record. And we cannot know that before reading
   the metadata BLOB, whose starting point we are
   trying to find here. (Although the PRIMARY KEY
   columns cannot be NULL, we would have to know
   where the lengths of variable-length PRIMARY KEY
   columns start.)

   So, unfortunately we cannot help users who
   were affected by MDEV-21088 on a ROW_FORMAT=COMPACT
or ROW_FORMAT=DYNAMIC table. */

   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
      || mach_read_from_4(ptr + BTR_EXTERN_SPACE_ID)
      != space->id) {
   goto incompatible;
  }

  buf_block_t* block = buf_page_get(
   page_id_t(space->id,
      mach_read_from_4(ptr + BTR_EXTERN_PAGE_NO)),
   0, RW_S_LATCH, mtr);
  if (!block) {
   goto incompatible;
  }

  btr_search_drop_page_hash_index(block, index);

  if (fil_page_get_type(block->page.frame) != FIL_PAGE_TYPE_BLOB
      || mach_read_from_4(&block->page.frame
     [FIL_PAGE_DATA
      + BTR_BLOB_HDR_NEXT_PAGE_NO])
      != FIL_NULL
      || mach_read_from_4(&block->page.frame
     [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 = block->page.frame
         + (FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE) + len,
         * const end = block->page.frame + srv_page_size
         - BTR_EXTERN_LEN;
       b < end; ) {
   if (*b++) {
    goto incompatible;
   }
  }

  if (index->table->deserialise_columns(
       &block->page.frame
       [FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE], len)) {
   goto incompatible;
  }

  /* Proceed to initialize the default values of
any instantly added columns. */
 }

 mem_heap_t* heap = NULL;
 rec_offs* offsets = rec_get_offsets(rec, index, NULL,
         index->n_core_fields,
         ULINT_UNDEFINED, &heap);
 if (rec_offs_any_default(offsets)) {
inconsistent:
  mem_heap_free(heap);
  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,
    cur.page_cur.block->zip_size(),
    len, index->table->heap);
  }
 }

 mem_heap_free(heap);
 return DB_SUCCESS;
}

/** Load the instant ALTER TABLE metadata from the clustered index
when loading a table definition.
@param[in,out] mtr mini-transaction
@param[in,out] table table definition from the data dictionary
@return error code
@retval DB_SUCCESS if no error occurred */
dberr_t btr_cur_instant_init(mtr_t *mtr, dict_table_t *table)
{
  dict_index_t *index= dict_table_get_first_index(table);
  mtr->start();
  dberr_t err= index ? btr_cur_instant_init_low(index, mtr) : DB_CORRUPTION;
  mtr->commit();
  if (err == DB_SUCCESS && index->is_gen_clust())
  {
    btr_cur_t cur;
    mtr->start();
    err= cur.open_leaf(false, index, BTR_SEARCH_LEAF, mtr);
    if (err != DB_SUCCESS);
    else if (const rec_t *rec= page_rec_get_prev(btr_cur_get_rec(&cur)))
      if (page_rec_is_user_rec(rec))
        table->row_id= mach_read_from_6(rec);
    mtr->commit();
  }
  return err;
}

/** Initialize the n_core_null_bytes on first access to a clustered
index root page.
@param[in] index clustered index that is on its first access
@param[in] page clustered index root page
@return whether the page is corrupted */
bool btr_cur_instant_root_init(dict_index_t* index, const page_t* page)
{
 ut_ad(!index->is_dummy);
 ut_ad(index->is_primary());
 ut_ad(!index->is_instant());
 ut_ad(index->table->supports_instant());

 if (page_has_siblings(page)) {
  return true;
 }

 /* This is normally executed as part of btr_cur_instant_init()
 when dict_load_table_one() is loading a table definition.
 Other threads should not access or modify the n_core_null_bytes,
 n_core_fields before dict_load_table_one() returns.

 This can also be executed during IMPORT TABLESPACE, where the
table definition is exclusively locked. */

 switch (fil_page_get_type(page)) {
 default:
  return true;
 case FIL_PAGE_INDEX:
  /* The field PAGE_INSTANT is guaranteed 0 on clustered
  index root pages of ROW_FORMAT=COMPACT or
ROW_FORMAT=DYNAMIC when instant ADD COLUMN is not used. */
  if (page_is_comp(page) && page_get_instant(page)) {
   return true;
  }
  index->n_core_null_bytes = static_cast<uint8_t>(
   UT_BITS_IN_BYTES(unsigned(index->n_nullable)));
  return false;
 case FIL_PAGE_TYPE_INSTANT:
  break;
 }

 const uint16_t n = page_get_instant(page);

 if (n < index->n_uniq + DATA_ROLL_PTR) {
  /* The PRIMARY KEY (or hidden DB_ROW_ID) and
  DB_TRX_ID,DB_ROLL_PTR columns must always be present
as 'core' fields. */
  return true;
 }

 if (n > REC_MAX_N_FIELDS) {
  return true;
 }

 index->n_core_fields = n & dict_index_t::MAX_N_FIELDS;

 const rec_t* infimum = page_get_infimum_rec(page);
 const rec_t* supremum = page_get_supremum_rec(page);

 if (!memcmp(infimum, "infimum", 8)
     && !memcmp(supremum, "supremum", 8)) {
  if (n > index->n_fields) {
   /* All fields, including those for instantly
   added columns, must be present in the
data dictionary. */
   return true;
  }

  ut_ad(!index->is_dummy);
  ut_d(index->is_dummy = true);
  index->n_core_null_bytes = static_cast<uint8_t>(
   UT_BITS_IN_BYTES(index->get_n_nullable(n)));
  ut_d(index->is_dummy = false);
  return false;
 }

 if (memcmp(infimum, field_ref_zero, 8)
     || memcmp(supremum, field_ref_zero, 7)) {
  /* The infimum and supremum records must either contain
  the original strings, or they must be filled with zero
bytes, except for the bytes that we have repurposed. */
  return true;
 }

 index->n_core_null_bytes = supremum[7];
 return index->n_core_null_bytes > 128;
}

/**
Gets intention in btr_intention_t from latch_mode, and cleares the intention
at the latch_mode.
@param latch_mode in/out: pointer to latch_mode
@return intention for latching tree */
static
btr_intention_t btr_cur_get_and_clear_intention(btr_latch_mode *latch_mode)
{
 btr_intention_t intention;

 switch (*latch_mode & (BTR_LATCH_FOR_INSERT | BTR_LATCH_FOR_DELETE)) {
 case BTR_LATCH_FOR_INSERT:
  intention = BTR_INTENTION_INSERT;
  break;
 case BTR_LATCH_FOR_DELETE:
  intention = BTR_INTENTION_DELETE;
  break;
 default:
  /* both or unknown */
  intention = BTR_INTENTION_BOTH;
 }
 *latch_mode = btr_latch_mode(
  *latch_mode & ~(BTR_LATCH_FOR_INSERT | BTR_LATCH_FOR_DELETE));

 return(intention);
}

/** @return whether the distance between two records is at most the
specified value */
template<bool comp>
static bool
page_rec_distance_is_at_most(const page_t *page, const rec_t *left,
                             const rec_t *right, ulint val)
  noexcept
{
  do
  {
    if (left == right)
      return true;
    left= page_rec_next_get<comp>(page, left);
  }
  while (left && val--);
  return false;
}

/** Detects whether the modifying record might need a modifying tree structure.
@param[in] index  index
@param[in] page  page
@param[in] lock_intention lock intention for the tree operation
@param[in] rec  record (current node_ptr)
@param[in] rec_size size of the record or max size of node_ptr
@param[in] zip_size ROW_FORMAT=COMPRESSED page size, or 0
@param[in] mtr  mtr
@return true if tree modification is needed */
static
bool
btr_cur_will_modify_tree(
 dict_index_t* index,
 const page_t* page,
 btr_intention_t lock_intention,
 const rec_t* rec,
 ulint  rec_size,
 ulint  zip_size,
 mtr_t*  mtr)
{
 ut_ad(!page_is_leaf(page));
 ut_ad(mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK
      | MTR_MEMO_SX_LOCK));

 /* Pessimistic delete of the first record causes delete & insert
 of node_ptr at upper level. And a subsequent page shrink is
 possible. It causes delete of node_ptr at the upper level.
 So we should pay attention also to 2nd record not only
 first record and last record. Because if the "delete & insert" are
 done for the different page, the 2nd record become
 first record and following compress might delete the record and causes
the uppper level node_ptr modification. */

 const ulint n_recs = page_get_n_recs(page);

 if (lock_intention <= BTR_INTENTION_BOTH) {
  compile_time_assert(BTR_INTENTION_DELETE < BTR_INTENTION_BOTH);
  compile_time_assert(BTR_INTENTION_BOTH < BTR_INTENTION_INSERT);

  if (!page_has_siblings(page)) {
   return true;
  }

  ulint margin = rec_size;

  if (lock_intention == BTR_INTENTION_BOTH) {
   ulint level = btr_page_get_level(page);

   /* This value is the worst expectation for the node_ptr
   records to be deleted from this page. It is used to
   expect whether the cursor position can be the left_most
record in this page or not. */
   ulint   max_nodes_deleted = 0;

   /* By modifying tree operations from the under of this
   level, logically (2 ^ (level - 1)) opportunities to
deleting records in maximum even unreally rare case. */
   if (level > 7) {
    /* TODO: adjust this practical limit. */
    max_nodes_deleted = 64;
   } else if (level > 0) {
    max_nodes_deleted = (ulint)1 << (level - 1);
   }
   /* check delete will cause. (BTR_INTENTION_BOTH
or BTR_INTENTION_DELETE) */
   if (n_recs <= max_nodes_deleted * 2) {
    /* The cursor record can be the left most record
in this page. */
    return true;
   }

   if (page_is_comp(page)) {
    const rec_t *const infimum
     = page + PAGE_NEW_INFIMUM;
    if (page_rec_next_get<true>(page, infimum)
        == rec) {
     return true;
    }
    if (page_has_prev(page)
        && page_rec_distance_is_at_most<true>(
         page, infimum, rec,
         max_nodes_deleted)) {
     return true;
    }
    if (page_has_next(page)
        && page_rec_distance_is_at_most<true>(
         page, rec,
         page + PAGE_NEW_SUPREMUM,
         max_nodes_deleted)) {
     return true;
    }
   } else {
    const rec_t *const infimum
     = page + PAGE_OLD_INFIMUM;
    if (page_rec_next_get<false>(page, infimum)
        == rec) {
     return true;
    }
    if (page_has_prev(page)
        && page_rec_distance_is_at_most<false>(
         page, infimum, rec,
         max_nodes_deleted)) {
     return true;
    }
    if (page_has_next(page)
        && page_rec_distance_is_at_most<false>(
         page, rec,
         page + PAGE_OLD_SUPREMUM,
         max_nodes_deleted)) {
     return true;
    }
   }

   /* Delete at leftmost record in a page causes delete
   & insert at its parent page. After that, the delete
   might cause btr_compress() and delete record at its
parent page. Thus we should consider max deletes. */
   margin *= max_nodes_deleted;
  }

  /* Safe because we already have SX latch of the index tree */
  if (page_get_data_size(page)
      < margin + BTR_CUR_PAGE_COMPRESS_LIMIT(index)) {
   return(true);
  }
 }

 if (lock_intention >= BTR_INTENTION_BOTH) {
  /* check insert will cause. BTR_INTENTION_BOTH
or BTR_INTENTION_INSERT*/

  /* Once we invoke the btr_cur_limit_optimistic_insert_debug,
  we should check it here in advance, since the max allowable
records in a page is limited. */
  LIMIT_OPTIMISTIC_INSERT_DEBUG(n_recs, return true);

  /* needs 2 records' space for the case the single split and
  insert cannot fit.
  page_get_max_insert_size_after_reorganize() includes space
for page directory already */
  ulint max_size
   = page_get_max_insert_size_after_reorganize(page, 2);

  if (max_size < BTR_CUR_PAGE_REORGANIZE_LIMIT + rec_size
      || max_size < rec_size * 2) {
   return(true);
  }

  /* TODO: optimize this condition for ROW_FORMAT=COMPRESSED.
  This is based on the worst case, and we could invoke
page_zip_available() on the block->page.zip. */
  /* needs 2 records' space also for worst compress rate. */
  if (zip_size
      && page_zip_empty_size(index->n_fields, zip_size)
      <= rec_size * 2 + page_get_data_size(page)
      + page_dir_calc_reserved_space(n_recs + 2)) {
   return(true);
  }
 }

 return(false);
}

/** Detects whether the modifying record might need a opposite modification
to the intention.
@param bpage             buffer pool page
@param is_clust          whether this is a clustered index
@param lock_intention    lock intention for the tree operation
@param node_ptr_max_size the maximum size of a node pointer
@param compress_limit    BTR_CUR_PAGE_COMPRESS_LIMIT(index)
@param rec               record (current node_ptr)
@return true if tree modification is needed */
static bool btr_cur_need_opposite_intention(const buf_page_t &bpage,
                                            bool is_clust,
                                            btr_intention_t lock_intention,
                                            ulint node_ptr_max_size,
                                            ulint compress_limit,
                                            const rec_t *rec)
{
  ut_ad(bpage.frame == page_align(rec));
  if (UNIV_LIKELY_NULL(bpage.zip.data) &&
      !page_zip_available(&bpage.zip, is_clust, node_ptr_max_size, 1))
    return true;
  const page_t *const page= bpage.frame;
  if (lock_intention != BTR_INTENTION_INSERT)
  {
    /* We compensate also for btr_cur_compress_recommendation() */
    if (!page_has_siblings(page) ||
        page_rec_is_first(rec, page) || page_rec_is_last(rec, page) ||
        page_get_data_size(page) < node_ptr_max_size + compress_limit)
      return true;
    if (lock_intention == BTR_INTENTION_DELETE)
      return false;
  }
  else if (page_has_next(page) && page_rec_is_last(rec, page))
    return true;
  LIMIT_OPTIMISTIC_INSERT_DEBUG(page_get_n_recs(page), return true);
  const ulint max_size= page_get_max_insert_size_after_reorganize(page, 2);
  return max_size < BTR_CUR_PAGE_REORGANIZE_LIMIT + node_ptr_max_size ||
    max_size < node_ptr_max_size * 2;
}

/**
@param[in] index b-tree
@return maximum size of a node pointer record in bytes */
static ulint btr_node_ptr_max_size(const dict_index_t* index)
{
 /* Each record has page_no, length of page_no and header. */
 ulint comp = dict_table_is_comp(index->table);
 ulint rec_max_size = comp
  ? REC_NODE_PTR_SIZE + 1 + REC_N_NEW_EXTRA_BYTES
  + UT_BITS_IN_BYTES(index->n_nullable)
  : REC_NODE_PTR_SIZE + 2 + REC_N_OLD_EXTRA_BYTES
  + 2 * index->n_fields;

 /* Compute the maximum possible record size. */
 for (ulint i = 0; i < dict_index_get_n_unique_in_tree(index); i++) {
  const dict_field_t* field
   = dict_index_get_nth_field(index, i);
  const dict_col_t* col
   = dict_field_get_col(field);
  ulint   field_max_size;
  ulint   field_ext_max_size;

  /* Determine the maximum length of the index field. */

  field_max_size = dict_col_get_fixed_size(col, comp);
  if (field_max_size && field->fixed_len) {
   /* dict_index_add_col() should guarantee this */
   ut_ad(!field->prefix_len
         || field->fixed_len == field->prefix_len);
   /* Fixed lengths are not encoded
in ROW_FORMAT=COMPACT. */
   rec_max_size += field_max_size;
   continue;
  }

  field_max_size = dict_col_get_max_size(col);
  if (UNIV_UNLIKELY(!field_max_size)) {
   switch (col->mtype) {
   case DATA_VARCHAR:
    if (!comp
        && (!strcmp(index->table->name.m_name,
      "SYS_FOREIGN")
     || !strcmp(index->table->name.m_name,
         "SYS_FOREIGN_COLS"))) {
     break;
    }
    /* fall through */
   case DATA_FIXBINARY:
   case DATA_BINARY:
   case DATA_VARMYSQL:
   case DATA_CHAR:
   case DATA_MYSQL:
    /* BINARY(0), VARBINARY(0),
    CHAR(0) and VARCHAR(0) are possible
    data type definitions in MariaDB.
    The InnoDB internal SQL parser maps
    CHAR to DATA_VARCHAR, so DATA_CHAR (or
    DATA_MYSQL) is only coming from the
MariaDB SQL layer. */
    if (comp) {
     /* Add a length byte, because
     fixed-length empty field are
     encoded as variable-length.
     For ROW_FORMAT=REDUNDANT,
     these bytes were added to
rec_max_size before this loop. */
     rec_max_size++;
    }
    continue;
   }

   /* SYS_FOREIGN.ID is defined as CHAR in the
   InnoDB internal SQL parser, which translates
   into the incorrect VARCHAR(0).  InnoDB does
   not enforce maximum lengths of columns, so
   that is why any data can be inserted in the
   first place.

   Likewise, SYS_FOREIGN.FOR_NAME,
   SYS_FOREIGN.REF_NAME, SYS_FOREIGN_COLS.ID, are
defined as CHAR, and also they are part of a key. */

   ut_ad(!strcmp(index->table->name.m_name,
          "SYS_FOREIGN")
         || !strcmp(index->table->name.m_name,
      "SYS_FOREIGN_COLS"));
   ut_ad(!comp);
   ut_ad(col->mtype == DATA_VARCHAR);

   rec_max_size += (srv_page_size == UNIV_PAGE_SIZE_MAX)
    ? REDUNDANT_REC_MAX_DATA_SIZE
    : page_get_free_space_of_empty(FALSE) / 2;
  } else if (field_max_size == NAME_LEN && i == 1
      && (!strcmp(index->table->name.m_name,
           TABLE_STATS_NAME)
          || !strcmp(index->table->name.m_name,
       INDEX_STATS_NAME))) {
   /* Interpret "table_name" as VARCHAR(199) even
   if it was incorrectly defined as VARCHAR(64).
   While the caller of ha_innobase enforces the
   maximum length on any data written, the InnoDB
   internal SQL parser will happily write as much
   data as is provided. The purpose of this hack
   is to avoid InnoDB hangs after persistent
   statistics on partitioned tables are
deleted. */
   field_max_size = 199 * SYSTEM_CHARSET_MBMAXLEN;
  }
  field_ext_max_size = field_max_size < 256 ? 1 : 2;

  if (field->prefix_len
      && field->prefix_len < field_max_size) {
   field_max_size = field->prefix_len;
  }

  if (comp) {
   /* Add the extra size for ROW_FORMAT=COMPACT.
   For ROW_FORMAT=REDUNDANT, these bytes were
added to rec_max_size before this loop. */
   rec_max_size += field_ext_max_size;
  }

  rec_max_size += field_max_size;
 }

 return rec_max_size;
}

/** @return a B-tree search mode suitable for non-leaf pages
@param mode  leaf page search mode */
static inline page_cur_mode_t btr_cur_nonleaf_mode(page_cur_mode_t mode)
{
  if (mode > PAGE_CUR_GE)
  {
    ut_ad(mode == PAGE_CUR_L || mode == PAGE_CUR_LE);
    return mode;
  }
  if (mode == PAGE_CUR_GE)
    return PAGE_CUR_L;
  ut_ad(mode == PAGE_CUR_G);
  return PAGE_CUR_LE;
}

MY_ATTRIBUTE((nonnull,warn_unused_result))
/** Acquire a latch on the previous page without violating the latching order.
@param rw_latch the latch on block (RW_S_LATCH or RW_X_LATCH)
@param page_id  page identifier with valid space identifier
@param err      error code
@param mtr      mini-transaction
@retval 0  if an error occurred
@retval 1  if the page could be latched in the wrong order
@retval -1 if the latch on block was temporarily released */
static int btr_latch_prev(rw_lock_type_t rw_latch,
                          page_id_t page_id, dberr_t *err, mtr_t *mtr) noexcept
{
  ut_ad(rw_latch == RW_S_LATCH || rw_latch == RW_X_LATCH);

  buf_block_t *block= mtr->at_savepoint(mtr->get_savepoint() - 1);

  ut_ad(page_id.space() == block->page.id().space());

  const page_t *const page= block->page.frame;
  page_id.set_page_no(btr_page_get_prev(page));
  /* We are holding a latch on the current page.

  We will start by buffer-fixing the left sibling. Waiting for a latch
  on it while holding a latch on the current page could lead to a
  deadlock, because another thread could hold that latch and wait for
  a right sibling page latch (the current page).

  If there is a conflict, we will temporarily release our latch on the
  current block while waiting for a latch on the left sibling.  The
  buffer-fixes on both blocks will prevent eviction. */

 retry:
  int ret= 1;
  buf_block_t *prev=
    buf_pool.page_fix(page_id, err, mtr->trx, buf_pool_t::FIX_NOWAIT);
  if (UNIV_UNLIKELY(!prev))
    return 0;
  if (prev == reinterpret_cast<buf_block_t*>(-1))
  {
    /* The block existed in buf_pool.page_hash, but not in a state that is
    safe to access without waiting for some pending operation, such as
    buf_page_t::read_complete() or buf_pool_t::unzip().

    Retry while temporarily releasing the successor block->page.lock
    (but retaining a buffer-fix so that the block cannot be evicted. */

    if (rw_latch == RW_S_LATCH)
      block->page.lock.s_unlock();
    else
      block->page.lock.x_unlock();

    prev= buf_pool.page_fix(page_id, err, mtr->trx, buf_pool_t::FIX_WAIT_READ);

    if (!prev)
    {
      ut_ad(*err != DB_SUCCESS);
      if (rw_latch == RW_S_LATCH)
        block->page.lock.s_lock();
      else
        block->page.lock.x_lock();
      return 0;
    }
    else if (rw_latch == RW_S_LATCH)
      goto wait_for_s;
    else
      goto wait_for_x;
  }

  static_assert(MTR_MEMO_PAGE_S_FIX == mtr_memo_type_t(BTR_SEARCH_LEAF), "");
  static_assert(MTR_MEMO_PAGE_X_FIX == mtr_memo_type_t(BTR_MODIFY_LEAF), "");

  if (rw_latch == RW_S_LATCH
      ? prev->page.lock.s_lock_try()
      : prev->page.lock.x_lock_try())
    mtr->memo_push(prev, mtr_memo_type_t(rw_latch));
  else
  {
    if (rw_latch == RW_S_LATCH)
    {
      block->page.lock.s_unlock();
    wait_for_s:
      prev->page.lock.s_lock();
      block->page.lock.s_lock();
    }
    else
    {
      block->page.lock.x_unlock();
    wait_for_x:
      prev->page.lock.x_lock();
      block->page.lock.x_lock();
    }

    ut_ad(block == mtr->at_savepoint(mtr->get_savepoint() - 1));
    mtr->memo_push(prev, mtr_memo_type_t(rw_latch));
    const page_id_t prev_page_id= page_id;
    page_id.set_page_no(btr_page_get_prev(page));
    ret= -1;

    if (UNIV_UNLIKELY(page_id != prev_page_id))
    {
      mtr->release_last_page();
      if (page_id.page_no() == FIL_NULL)
        return ret;
      goto retry;
    }
  }

  const page_t *const p= prev->page.frame;
  if (memcmp_aligned<4>(FIL_PAGE_NEXT + p, FIL_PAGE_OFFSET + page, 4) ||
      memcmp_aligned<2>(FIL_PAGE_TYPE + p, FIL_PAGE_TYPE + page, 2) ||
      memcmp_aligned<2>(PAGE_HEADER + PAGE_INDEX_ID + p,
                        PAGE_HEADER + PAGE_INDEX_ID + page, 8) ||
      page_is_comp(p) != page_is_comp(page))
  {
    ut_ad("corrupted" == 0); // FIXME: remove this
    *err= DB_CORRUPTION;
    ret= 0;
  }

  return ret;
}

#ifdef BTR_CUR_HASH_ADAPT
/** Increment successful adaptive hash index lookups */
static inline void btr_ahi_inc_searches(const mtr_t &mtr) noexcept
{
  mtr.trx->n_sea++;
  if (ha_handler_stats *stats= mtr.trx->active_handler_stats)
    stats->ahi_searches++;
}

/** Increment adaptive hash index misses (B-tree fallback) */
static inline void btr_ahi_inc_searches_btree(const mtr_t &mtr) noexcept
{
  mtr.trx->n_non_sea++;
  if (ha_handler_stats *stats= mtr.trx->active_handler_stats)
    stats->ahi_searches_btree++;
}
#endif /* BTR_CUR_HASH_ADAPT */

dberr_t btr_cur_t::search_leaf(const dtuple_t *tuple, page_cur_mode_t mode,
                               btr_latch_mode latch_mode, mtr_t *mtr)
{
  ut_ad(index()->is_btree());

  buf_block_t *guess;
  btr_intention_t lock_intention;
  bool detected_same_key_root= false;

  mem_heap_t *heap= nullptr;
  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets= offsets_;
  rec_offs offsets2_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets2= offsets2_;
  rec_offs_init(offsets_);
  rec_offs_init(offsets2_);

  ut_ad(dict_index_check_search_tuple(index(), tuple));
  ut_ad(dtuple_check_typed(tuple));
  ut_ad(index()->page != FIL_NULL);

  MEM_UNDEFINED(&up_match, sizeof up_match);
  MEM_UNDEFINED(&up_bytes, sizeof up_bytes);
  MEM_UNDEFINED(&low_match, sizeof low_match);
  MEM_UNDEFINED(&low_bytes, sizeof low_bytes);
  ut_d(up_match= low_match= uint16_t(~0u));

  ut_ad(!(latch_mode & BTR_ALREADY_S_LATCHED) ||
        mtr->memo_contains_flagged(&index()->lock,
                                   MTR_MEMO_S_LOCK | MTR_MEMO_SX_LOCK |
                                   MTR_MEMO_X_LOCK));

  const bool latch_by_caller= latch_mode & BTR_ALREADY_S_LATCHED;
  lock_intention= btr_cur_get_and_clear_intention(&latch_mode);
  latch_mode= BTR_LATCH_MODE_WITHOUT_FLAGS(latch_mode);

  ut_ad(!latch_by_caller
        || latch_mode == BTR_SEARCH_LEAF
        || latch_mode == BTR_MODIFY_LEAF
        || latch_mode == BTR_MODIFY_TREE
        || latch_mode == BTR_MODIFY_ROOT_AND_LEAF);

#ifndef BTR_CUR_ADAPT
  guess= nullptr;
#else
  auto info= &index()->search_info;
  guess= info->root_guess;

# ifdef BTR_CUR_HASH_ADAPT
  flag= BTR_CUR_BINARY;
#  ifdef UNIV_SEARCH_PERF_STAT
  info->n_searches++;
#  endif
  if (latch_mode > BTR_MODIFY_LEAF)
    /* The adaptive hash index cannot be useful for these searches. */;
  else if (mode != PAGE_CUR_LE && mode != PAGE_CUR_GE)
    ut_ad(mode == PAGE_CUR_L || mode == PAGE_CUR_G);
  /* We do a dirty read of btr_search.enabled below,
  and btr_search_guess_on_hash() will have to check it again. */
  else if (!btr_search.is_enabled(index())) ;
  else if (btr_search_guess_on_hash(index(), tuple, mode != PAGE_CUR_LE,
                                    latch_mode, this, mtr))
  {
    /* Search using the hash index succeeded */
    ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_GE);
    ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_LE);
    ut_ad(low_match != uint16_t(~0U) || mode != PAGE_CUR_LE);
    DBUG_EXECUTE_IF("btr_cur_n_sea_delay", my_sleep(200000););
    btr_ahi_inc_searches(*mtr);

    return DB_SUCCESS;
  }
  else
    btr_ahi_inc_searches_btree(*mtr);
# endif
#endif

  /* If the hash search did not succeed, do binary search down the
     tree */

  /* Store the position of the tree latch we push to mtr so that we
     know how to release it when we have latched leaf node(s) */

  const ulint savepoint= mtr->get_savepoint();

  ulint node_ptr_max_size= 0, compress_limit= 0;
  rw_lock_type_t rw_latch= RW_S_LATCH;

  switch (latch_mode) {
  case BTR_MODIFY_TREE:
    rw_latch= RW_X_LATCH;
    node_ptr_max_size= btr_node_ptr_max_size(index());
    if (latch_by_caller)
    {
      ut_ad(mtr->memo_contains_flagged(&index()->lock, MTR_MEMO_X_LOCK));
      break;
    }
    if (lock_intention == BTR_INTENTION_DELETE)
    {
      compress_limit= BTR_CUR_PAGE_COMPRESS_LIMIT(index());
      if (os_aio_pending_reads_approx() &&
          trx_sys.history_size_approx() > BTR_CUR_FINE_HISTORY_LENGTH)
      {
        /* Most delete-intended operations are due to the purge of history.
        Prioritize them when the history list is growing huge. */
        mtr_x_lock_index(index(), mtr);
        break;
      }
    }
    mtr_sx_lock_index(index(), mtr);
    break;
#ifdef UNIV_DEBUG
  case BTR_CONT_MODIFY_TREE:
    ut_ad("invalid mode" == 0);
    break;
#endif
  case BTR_MODIFY_ROOT_AND_LEAF:
    rw_latch= RW_SX_LATCH;
    /* fall through */
  default:
    if (!latch_by_caller)
      mtr_s_lock_index(index(), mtr);
  }

  dberr_t err;

  if (!index()->table->space)
  {
  corrupted:
    ut_ad("corrupted" == 0); // FIXME: remove this
    err= DB_CORRUPTION;
  func_exit:
    if (UNIV_LIKELY_NULL(heap))
      mem_heap_free(heap);
    return err;
  }

  const ulint zip_size= index()->table->space->zip_size();

  /* Start with the root page. */
  page_id_t page_id(index()->table->space_id, index()->page);

  const page_cur_mode_t page_mode= btr_cur_nonleaf_mode(mode);
  ulint height= ULINT_UNDEFINED;
  up_match= 0;
  up_bytes= 0;
  low_match= 0;
  low_bytes= 0;
 search_loop:
  auto block_savepoint= mtr->get_savepoint();
  buf_block_t *block=
    buf_page_get_gen(page_id, zip_size, rw_latch, guess, BUF_GET, mtr, &err);
  if (!block)
  {
    btr_read_failed(err, *index());
    goto func_exit;
  }

  btr_search_drop_page_hash_index(block, index());

  if (!!page_is_comp(block->page.frame) != index()->table->not_redundant() ||
      btr_page_get_index_id(block->page.frame) != index()->id ||
      fil_page_get_type(block->page.frame) == FIL_PAGE_RTREE ||
      !fil_page_index_page_check(block->page.frame))
    goto corrupted;

  page_cur.block= block;
  ut_ad(block == mtr->at_savepoint(block_savepoint));
  const bool not_first_access{buf_page_make_young_if_needed(&block->page)};
#ifdef UNIV_ZIP_DEBUG
  if (const page_zip_des_t *page_zip= buf_block_get_page_zip(block))
    ut_a(page_zip_validate(page_zip, block->page.frame, index()));
#endif /* UNIV_ZIP_DEBUG */

  uint32_t page_level= btr_page_get_level(block->page.frame);

  if (height == ULINT_UNDEFINED)
  {
    /* We are in the B-tree index root page. */
#ifdef BTR_CUR_ADAPT
    info->root_guess= block;
#endif
  reached_root:
    height= page_level;
    tree_height= height + 1;

    if (!height)
    {
      /* The root page is also a leaf page.
      We may have to reacquire the page latch in a different mode. */
      switch (rw_latch) {
      case RW_S_LATCH:
        if (!(latch_mode & BTR_SEARCH_LEAF))
        {
          rw_latch= RW_X_LATCH;
          ut_ad(rw_lock_type_t(latch_mode & ~12) == RW_X_LATCH);
          mtr->lock_register(block_savepoint, MTR_MEMO_PAGE_X_FIX);
          if (!block->page.lock.s_x_upgrade_try())
          {
            block->page.lock.s_unlock();
            block->page.lock.x_lock();
            /* Dropping the index tree (and freeing the root page)
            should be impossible while we hold index()->lock. */
            ut_ad(!block->page.is_freed());
            page_level= btr_page_get_level(block->page.frame);
            if (UNIV_UNLIKELY(page_level != 0))
            {
              /* btr_root_raise_and_insert() was executed meanwhile */
              ut_ad(mtr->memo_contains_flagged(&index()->lock,
                                               MTR_MEMO_S_LOCK));
              block->page.lock.x_u_downgrade();
              block->page.lock.u_s_downgrade();
              rw_latch= RW_S_LATCH;
              mtr->lock_register(block_savepoint, MTR_MEMO_PAGE_S_FIX);
              goto reached_root;
            }
          }
        }
        if (rw_latch != RW_S_LATCH)
          break;
        if (!latch_by_caller)
          /* Release the tree s-latch */
          mtr->rollback_to_savepoint(savepoint, savepoint + 1);
        goto reached_latched_leaf;
      case RW_SX_LATCH:
        ut_ad(latch_mode == BTR_MODIFY_ROOT_AND_LEAF);
        static_assert(int{BTR_MODIFY_ROOT_AND_LEAF} == int{RW_SX_LATCH}, "");
        rw_latch= RW_X_LATCH;
        mtr->lock_register(block_savepoint, MTR_MEMO_PAGE_X_FIX);
        block->page.lock.u_x_upgrade();
        break;
      case RW_X_LATCH:
        if (latch_mode == BTR_MODIFY_TREE)
          goto reached_index_root_and_leaf;
        break;
      case RW_NO_LATCH:
        ut_ad(0);
      }
      goto reached_root_and_leaf;
    }
  }
  else if (UNIV_UNLIKELY(height != page_level))
    goto corrupted;
  else
    switch (latch_mode) {
    case BTR_MODIFY_TREE:
      break;
    case BTR_MODIFY_ROOT_AND_LEAF:
      ut_ad((mtr->at_savepoint(block_savepoint - 1)->page.id().page_no() ==
             index()->page) == (tree_height <= height + 2));
      if (tree_height <= height + 2)
        /* Retain the root page latch. */
        break;
      /* fall through */
    default:
      ut_ad(block_savepoint > savepoint);
      mtr->rollback_to_savepoint(block_savepoint - 1, block_savepoint);
      block_savepoint--;
    }

  if (!height)
  {
    /* We reached the leaf level. */
    ut_ad(block == mtr->at_savepoint(block_savepoint));

    if (latch_mode == BTR_MODIFY_ROOT_AND_LEAF)
    {
    reached_root_and_leaf:
      if (!latch_by_caller)
        mtr->rollback_to_savepoint(savepoint, savepoint + 1);
    reached_index_root_and_leaf:
      ut_ad(rw_latch == RW_X_LATCH);
      btr_search_drop_page_hash_index(block, index());
      if (page_cur_search_with_match(tuple, mode, &up_match, &low_match,
                                     &page_cur, nullptr))
        goto corrupted;
      ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_GE);
      ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_LE);
      ut_ad(low_match != uint16_t(~0U) || mode != PAGE_CUR_LE);
      goto func_exit;
    }

    switch (latch_mode) {
    case BTR_SEARCH_PREV: /* btr_pcur_move_to_prev() */
      ut_ad(!latch_by_caller);
      ut_ad(rw_latch == RW_S_LATCH);

      /* latch also siblings from left to right */
      if (page_has_prev(block->page.frame) &&
          !btr_latch_prev(rw_latch, page_id, &err, mtr))
        goto func_exit;
      if (page_has_next(block->page.frame) &&
          !btr_block_get(*index(), btr_page_get_next(block->page.frame),
                         rw_latch, mtr, &err))
        goto func_exit;
      goto release_tree;
    case BTR_SEARCH_LEAF:
    case BTR_MODIFY_LEAF:
      ut_ad(rw_latch == rw_lock_type_t(latch_mode));
      if (!latch_by_caller)
      {
release_tree:
        /* Release the tree s-latch */
        block_savepoint--;
        mtr->rollback_to_savepoint(savepoint, savepoint + 1);
      }
      /* release upper blocks */
      if (savepoint < block_savepoint)
        mtr->rollback_to_savepoint(savepoint, block_savepoint);
      break;
    default:
      ut_ad(latch_mode == BTR_MODIFY_TREE);
      ut_ad(rw_latch == RW_X_LATCH);
      /* x-latch also siblings from left to right */
      if (page_has_prev(block->page.frame) &&
          !btr_latch_prev(rw_latch, page_id, &err, mtr))
        goto func_exit;
      if (page_has_next(block->page.frame) &&
          !btr_block_get(*index(), btr_page_get_next(block->page.frame),
                         RW_X_LATCH, mtr, &err))
        goto func_exit;
    }

  reached_latched_leaf:
    if (!(tuple->info_bits & REC_INFO_MIN_REC_FLAG))
    {
      if (page_cur_search_with_match_bytes(*tuple, mode, &up_match, &low_match,
                                           &page_cur, &up_bytes, &low_bytes))
        goto corrupted;
    }
    else if (page_cur_search_with_match(tuple, mode, &up_match, &low_match,
                                        &page_cur, nullptr))
      goto corrupted;

    ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_GE);
    ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_LE);
    ut_ad(low_match != uint16_t(~0U) || mode != PAGE_CUR_LE);

    if (latch_mode == BTR_MODIFY_TREE &&
        btr_cur_need_opposite_intention(block->page, index()->is_clust(),
                                        lock_intention,
                                        node_ptr_max_size, compress_limit,
                                        page_cur.rec))
        goto need_opposite_intention;

#ifdef BTR_CUR_HASH_ADAPT
    if (flag != BTR_CUR_BINARY)
    {
      ut_ad(!(tuple->info_bits & REC_INFO_MIN_REC_FLAG));
      ut_ad(!index()->table->is_temporary());
      if (!rec_is_metadata(page_cur.rec, *index()) &&
          index()->search_info.hash_analysis_useful())
        search_info_update(*mtr);
    }
#endif /* BTR_CUR_HASH_ADAPT */

    goto func_exit;
  }

  guess= nullptr;
  if (page_cur_search_with_match(tuple, page_mode, &up_match, &low_match,
                                 &page_cur, nullptr))
    goto corrupted;
  offsets= rec_get_offsets(page_cur.rec, index(), offsets, 0, ULINT_UNDEFINED,
                           &heap);

  ut_ad(block == mtr->at_savepoint(block_savepoint));

  switch (latch_mode) {
  default:
    break;
  case BTR_MODIFY_TREE:
    if (btr_cur_need_opposite_intention(block->page, index()->is_clust(),
                                        lock_intention,
                                        node_ptr_max_size, compress_limit,
                                        page_cur.rec))
      /* If the rec is the first or last in the page for pessimistic
      delete intention, it might cause node_ptr insert for the upper
      level. We should change the intention and retry. */
    need_opposite_intention:
      return pessimistic_search_leaf(tuple, mode, mtr);

    if (detected_same_key_root || lock_intention != BTR_INTENTION_BOTH ||
        index()->is_unique() ||
        (up_match <= rec_offs_n_fields(offsets) &&
         low_match <= rec_offs_n_fields(offsets)))
      break;

    /* If the first or the last record of the page or the same key
    value to the first record or last record, then another page might
    be chosen when BTR_CONT_MODIFY_TREE.  So, the parent page should
    not released to avoiding deadlock with blocking the another search
    with the same key value. */
    const rec_t *first=
      page_rec_get_next_const(page_get_infimum_rec(block->page.frame));
    ulint matched_fields;

    if (UNIV_UNLIKELY(!first))
      goto corrupted;
    if (page_cur.rec == first ||
        page_rec_is_last(page_cur.rec, block->page.frame))
    {
    same_key_root:
      detected_same_key_root= true;
      break;
    }

    matched_fields= 0;
    offsets2= rec_get_offsets(first, index(), offsets2, 0, ULINT_UNDEFINED,
                              &heap);
    cmp_rec_rec(page_cur.rec, first, offsets, offsets2, index(), false,
                &matched_fields);
    if (matched_fields >= rec_offs_n_fields(offsets) - 1)
      goto same_key_root;
    if (const rec_t* last=
        page_rec_get_prev_const(page_get_supremum_rec(block->page.frame)))
    {
      matched_fields= 0;
      offsets2= rec_get_offsets(last, index(), offsets2, 0, ULINT_UNDEFINED,
                                &heap);
      cmp_rec_rec(page_cur.rec, last, offsets, offsets2, index(), false,
                  &matched_fields);
      if (matched_fields >= rec_offs_n_fields(offsets) - 1)
        goto same_key_root;
    }
    else
      goto corrupted;

    /* Release the non-root parent page unless it may need to be modified. */
    if (tree_height > height + 1 &&
        !btr_cur_will_modify_tree(index(), block->page.frame, lock_intention,
                                  page_cur.rec, node_ptr_max_size,
                                  zip_size, mtr))
    {
      mtr->rollback_to_savepoint(block_savepoint - 1, block_savepoint);
      block_savepoint--;
    }
  }

  /* Go to the child node */
  page_id.set_page_no(btr_node_ptr_get_child_page_no(page_cur.rec, offsets));

  if (!--height)
  {
    /* We are about to access the leaf level. */

    switch (latch_mode) {
    case BTR_MODIFY_ROOT_AND_LEAF:
      rw_latch= RW_X_LATCH;
      break;
    case BTR_SEARCH_PREV: /* btr_pcur_move_to_prev() */
      ut_ad(rw_latch == RW_S_LATCH);

      if (!not_first_access)
        buf_read_ahead_linear(page_id);

      if (page_has_prev(block->page.frame) &&
          page_rec_is_first(page_cur.rec, block->page.frame))
      {
        ut_ad(block_savepoint + 1 == mtr->get_savepoint());

        /* Latch the previous page if the node pointer is the leftmost
        of the current page. */
        int ret= btr_latch_prev(rw_latch, page_id, &err, mtr);
        if (!ret)
          goto func_exit;
        ut_ad(block_savepoint + 2 == mtr->get_savepoint());
        if (ret < 0)
        {
          up_match= 0, low_match= 0, up_bytes= 0, low_bytes= 0;
          /* While our latch on the level-2 page prevents splits or
          merges of this level-1 block, other threads may have
          modified it due to splitting or merging some level-0 (leaf)
          pages underneath it. Thus, we must search again. */
          if (page_cur_search_with_match(tuple, page_mode,
                                         &up_match, &low_match,
                                         &page_cur, nullptr))
            goto corrupted;
          offsets= rec_get_offsets(page_cur.rec, index(), offsets, 0,
                                   ULINT_UNDEFINED, &heap);
          page_id.set_page_no(btr_node_ptr_get_child_page_no(page_cur.rec,
                                                             offsets));
        }
      }
      rw_latch= rw_lock_type_t(latch_mode & (RW_X_LATCH | RW_S_LATCH));
      break;
    case BTR_MODIFY_LEAF:
    case BTR_SEARCH_LEAF:
      rw_latch= rw_lock_type_t(latch_mode);
      if (!not_first_access)
        buf_read_ahead_linear(page_id);
      break;
    case BTR_MODIFY_TREE:
      ut_ad(rw_latch == RW_X_LATCH);

      if (lock_intention == BTR_INTENTION_INSERT &&
          page_has_next(block->page.frame) &&
          page_rec_is_last(page_cur.rec, block->page.frame))
      {
        /* btr_insert_into_right_sibling() might cause deleting node_ptr
        at upper level */
        mtr->rollback_to_savepoint(block_savepoint);
        goto need_opposite_intention;
      }
      break;
    default:
      ut_ad(rw_latch == RW_X_LATCH);
    }
  }

  goto search_loop;
}

ATTRIBUTE_COLD void mtr_t::index_lock_upgrade()
{
  auto &slot= m_memo[get_savepoint() - 1];
  if (slot.type == MTR_MEMO_X_LOCK)
    return;
  ut_ad(slot.type == MTR_MEMO_SX_LOCK);
  index_lock *lock= static_cast<index_lock*>(slot.object);
  lock->u_x_upgrade(SRW_LOCK_CALL);
  slot.type= MTR_MEMO_X_LOCK;
  ut_d(++btr_cur_n_index_lock_upgrades);
}

/** Mark a non-leaf page "least recently used", but avoid invoking
buf_page_t::set_accessed(), because we do not want linear read-ahead */
static void btr_cur_nonleaf_make_young(buf_page_t *bpage)
{
  if (UNIV_UNLIKELY(buf_page_peek_if_too_old(bpage)))
    buf_page_make_young(bpage);
}

ATTRIBUTE_COLD
dberr_t btr_cur_t::pessimistic_search_leaf(const dtuple_t *tuple,
                                           page_cur_mode_t mode, mtr_t *mtr)
{
  ut_ad(index()->is_btree());

  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs* offsets= offsets_;
  rec_offs_init(offsets_);

  ut_ad(flag == BTR_CUR_BINARY);
  ut_ad(dict_index_check_search_tuple(index(), tuple));
  ut_ad(dtuple_check_typed(tuple));
  buf_block_t *block= mtr->at_savepoint(1);
  ut_ad(block->page.id().page_no() == index()->page);
  block->page.fix();
  mtr->rollback_to_savepoint(1);
  mtr->index_lock_upgrade();

  const page_cur_mode_t page_mode{btr_cur_nonleaf_mode(mode)};

  mtr->page_lock(block, RW_X_LATCH);
  btr_search_drop_page_hash_index(block, index());

  up_match= 0;
  up_bytes= 0;
  low_match= 0;
  low_bytes= 0;
  ulint height= btr_page_get_level(block->page.frame);
  tree_height= height + 1;
  mem_heap_t *heap= nullptr;

 search_loop:
  dberr_t err;
  page_cur.block= block;

  if (UNIV_UNLIKELY(!height))
  {
    if (page_cur_search_with_match(tuple, mode, &up_match, &low_match,
                                   &page_cur, nullptr))
    corrupted:
      err= DB_CORRUPTION;
    else
    {
      ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_GE);
      ut_ad(up_match != uint16_t(~0U) || mode != PAGE_CUR_LE);
      ut_ad(low_match != uint16_t(~0U) || mode != PAGE_CUR_LE);

#ifdef BTR_CUR_HASH_ADAPT
      /* We do a dirty read of btr_search.enabled here.  We will recheck in
      btr_search_build_page_hash_index() before building a page hash
      index, while holding search latch. */
      if (!btr_search.is_enabled(index())) ;
      else if (tuple->info_bits & REC_INFO_MIN_REC_FLAG)
        /* This may be a search tuple for btr_pcur_t::restore_position(). */
        ut_ad(tuple->is_metadata() ||
              (tuple->is_metadata(tuple->info_bits ^ REC_STATUS_INSTANT)));
      else if (index()->table->is_temporary());
      else if (!rec_is_metadata(page_cur.rec, *index()) &&
               index()->search_info.hash_analysis_useful())
        search_info_update(*mtr);
#endif /* BTR_CUR_HASH_ADAPT */
      err= DB_SUCCESS;
    }

  func_exit:
    if (UNIV_LIKELY_NULL(heap))
      mem_heap_free(heap);
    return err;
  }

  if (page_cur_search_with_match(tuple, page_mode, &up_match, &low_match,
                                 &page_cur, nullptr))
    goto corrupted;

  page_id_t page_id{block->page.id()};

  offsets= rec_get_offsets(page_cur.rec, index(), offsets, 0, ULINT_UNDEFINED,
                           &heap);
  /* Go to the child node */
  page_id.set_page_no(btr_node_ptr_get_child_page_no(page_cur.rec, offsets));

  block=
    buf_page_get_gen(page_id, block->zip_size(), RW_X_LATCH, nullptr, BUF_GET,
                     mtr, &err);

  if (!block)
  {
    btr_read_failed(err, *index());
    goto func_exit;
  }

  btr_search_drop_page_hash_index(block, index());

  if (!!page_is_comp(block->page.frame) != index()->table->not_redundant() ||
      btr_page_get_index_id(block->page.frame) != index()->id ||
      fil_page_get_type(block->page.frame) == FIL_PAGE_RTREE ||
      !fil_page_index_page_check(block->page.frame))
    goto corrupted;

  if (--height != btr_page_get_level(block->page.frame))
    goto corrupted;

  btr_cur_nonleaf_make_young(&block->page);

#ifdef UNIV_ZIP_DEBUG
  const page_zip_des_t *page_zip= buf_block_get_page_zip(block);
  ut_a(!page_zip || page_zip_validate(page_zip, block->page.frame, index()));
#endif /* UNIV_ZIP_DEBUG */

  if (page_has_prev(block->page.frame) &&
      !btr_latch_prev(RW_X_LATCH, page_id, &err, mtr))
    goto func_exit;
  if (page_has_next(block->page.frame) &&
      !btr_block_get(*index(), btr_page_get_next(block->page.frame),
                     RW_X_LATCH, mtr, &err))
    goto func_exit;
  goto search_loop;
}

/********************************************************************//**
Searches an index tree and positions a tree cursor on a given non-leaf level.
NOTE: n_fields_cmp in tuple must be set so that it cannot be compared
to node pointer page number fields on the upper levels of the tree!
cursor->up_match and cursor->low_match both will have sensible values.
Cursor is left at the place where an insert of the
search tuple should be performed in the B-tree. InnoDB does an insert
immediately after the cursor. Thus, the cursor may end up on a user record,
or on a page infimum record.
@param level      the tree level of search
@param tuple      data tuple; NOTE: n_fields_cmp in tuple must be set so that
                  it cannot get compared to the node ptr page number field!
@param latch      RW_S_LATCH or RW_X_LATCH
@param cursor     tree cursor; the cursor page is s- or x-latched, but see also
                  above!
@param mtr        mini-transaction
@return DB_SUCCESS on success or error code otherwise */
dberr_t btr_cur_search_to_nth_level(ulint level,
                                    const dtuple_t *tuple,
                                    rw_lock_type_t rw_latch,
                                    btr_cur_t *cursor, mtr_t *mtr)
{
  dict_index_t *const index= cursor->index();

  ut_ad(index->is_btree());
  mem_heap_t *heap= nullptr;
  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets= offsets_;
  rec_offs_init(offsets_);
  ut_ad(level);
  ut_ad(dict_index_check_search_tuple(index, tuple));
  ut_ad(index->is_btree());
  ut_ad(dtuple_check_typed(tuple));
  ut_ad(index->page != FIL_NULL);

  MEM_UNDEFINED(&cursor->up_bytes, sizeof cursor->up_bytes);
  MEM_UNDEFINED(&cursor->low_bytes, sizeof cursor->low_bytes);
  cursor->up_match= 0;
  cursor->low_match= 0;
#ifdef BTR_CUR_HASH_ADAPT
  cursor->flag= BTR_CUR_BINARY;
#endif
#ifndef BTR_CUR_ADAPT
  buf_block_t *block= nullptr;
#else
  buf_block_t *block= index->search_info.root_guess;
#endif /* BTR_CUR_ADAPT */

  ut_ad(mtr->memo_contains_flagged(&index->lock,
                                   MTR_MEMO_X_LOCK | MTR_MEMO_SX_LOCK));

  dberr_t err;

  if (!index->table->space)
  {
  corrupted:
    err= DB_CORRUPTION;
  func_exit:
    if (UNIV_LIKELY_NULL(heap))
      mem_heap_free(heap);
    return err;
  }

  const ulint zip_size= index->table->space->zip_size();

  /* Start with the root page. */
  page_id_t page_id(index->table->space_id, index->page);
  ulint height= ULINT_UNDEFINED;

search_loop:
  err= DB_SUCCESS;
  if (buf_block_t *b=
      mtr->get_already_latched(page_id, mtr_memo_type_t(rw_latch)))
    block= b;
  else if (!(block= buf_page_get_gen(page_id, zip_size, rw_latch,
                                     block, BUF_GET, mtr, &err)))
  {
    btr_read_failed(err, *index);
    goto func_exit;
  }
  else
  {
    btr_search_drop_page_hash_index(block, index);
    btr_cur_nonleaf_make_young(&block->page);
  }

#ifdef UNIV_ZIP_DEBUG
  if (const page_zip_des_t *page_zip= buf_block_get_page_zip(block))
    ut_a(page_zip_validate(page_zip, block->page.frame, index));
#endif /* UNIV_ZIP_DEBUG */

  if (!!page_is_comp(block->page.frame) != index->table->not_redundant() ||
      btr_page_get_index_id(block->page.frame) != index->id ||
      fil_page_get_type(block->page.frame) == FIL_PAGE_RTREE ||
      !fil_page_index_page_check(block->page.frame))
    goto corrupted;

  const uint32_t page_level= btr_page_get_level(block->page.frame);

  if (height == ULINT_UNDEFINED)
  {
    /* We are in the root node */
    height= page_level;
    if (!height)
      goto corrupted;
    cursor->tree_height= height + 1;
  }
  else if (height != ulint{page_level})
    goto corrupted;

  cursor->page_cur.block= block;

  /* Search for complete index fields. */
  if (page_cur_search_with_match(tuple, PAGE_CUR_LE, &cursor->up_match,
                                 &cursor->low_match, &cursor->page_cur,
                                 nullptr))
    goto corrupted;

  /* If this is the desired level, leave the loop */
  if (level == height)
    goto func_exit;

  ut_ad(height > level);
  height--;

  offsets = rec_get_offsets(cursor->page_cur.rec, index, offsets, 0,
                            ULINT_UNDEFINED, &heap);
  /* Go to the child node */
  page_id.set_page_no(btr_node_ptr_get_child_page_no(cursor->page_cur.rec,
                                                     offsets));
  block= nullptr;
  goto search_loop;
}

dberr_t btr_cur_t::open_leaf(bool first, dict_index_t *index,
                             btr_latch_mode latch_mode, mtr_t *mtr)
{
  ulint n_blocks= 0;
  mem_heap_t *heap= nullptr;
  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets= offsets_;
  dberr_t err;

  rec_offs_init(offsets_);

  const bool latch_by_caller= latch_mode & BTR_ALREADY_S_LATCHED;
  latch_mode= btr_latch_mode(latch_mode & ~BTR_ALREADY_S_LATCHED);

  btr_intention_t lock_intention= btr_cur_get_and_clear_intention(&latch_mode);

  /* Store the position of the tree latch we push to mtr so that we
  know how to release it when we have latched the leaf node */

  auto savepoint= mtr->get_savepoint();

  rw_lock_type_t upper_rw_latch= RW_X_LATCH;
  ulint node_ptr_max_size= 0, compress_limit= 0;

  if (latch_mode == BTR_MODIFY_TREE)
  {
    node_ptr_max_size= btr_node_ptr_max_size(index);
    /* Most of delete-intended operations are purging. Free blocks
    and read IO bandwidth should be prioritized for them, when the
    history list is growing huge. */
    savepoint++;
    if (lock_intention == BTR_INTENTION_DELETE)
    {
      compress_limit= BTR_CUR_PAGE_COMPRESS_LIMIT(index);

      if (os_aio_pending_reads_approx() &&
          trx_sys.history_size_approx() > BTR_CUR_FINE_HISTORY_LENGTH)
      {
        mtr_x_lock_index(index, mtr);
        goto index_locked;
      }
    }
    mtr_sx_lock_index(index, mtr);
  }
  else
  {
    static_assert(int{BTR_CONT_MODIFY_TREE} == (12 | BTR_MODIFY_LEAF), "");
    ut_ad(!(latch_mode & 8));
    /* This function doesn't need to lock left page of the leaf page */
    static_assert(int{BTR_SEARCH_PREV} == (4 | BTR_SEARCH_LEAF), "");
    latch_mode= btr_latch_mode(latch_mode & (RW_S_LATCH | RW_X_LATCH));
    ut_ad(!latch_by_caller ||
          mtr->memo_contains_flagged(&index->lock,
                                     MTR_MEMO_SX_LOCK | MTR_MEMO_S_LOCK));
    upper_rw_latch= RW_S_LATCH;
    if (!latch_by_caller)
    {
      savepoint++;
      mtr_s_lock_index(index, mtr);
    }
  }

index_locked:
  ut_ad(savepoint == mtr->get_savepoint());

  const rw_lock_type_t root_leaf_rw_latch=
    rw_lock_type_t(latch_mode & (RW_S_LATCH | RW_X_LATCH));

  page_cur.index = index;

  uint32_t page= index->page;

  for (ulint height= ULINT_UNDEFINED;;)
  {
    ut_ad(n_blocks < BTR_MAX_LEVELS);
    ut_ad(savepoint + n_blocks == mtr->get_savepoint());

    bool first_access= false;
    buf_block_t* block=
      btr_block_get(*index, page,
                    height ? upper_rw_latch : root_leaf_rw_latch,
                    mtr, &err, &first_access);
    ut_ad(!block == (err != DB_SUCCESS));

    if (!block)
      break;

    if (first)
      page_cur_set_before_first(block, &page_cur);
    else
      page_cur_set_after_last(block, &page_cur);

    const uint32_t l= btr_page_get_level(block->page.frame);

    if (height == ULINT_UNDEFINED)
    {
      /* We are in the root node */
      height= l;
      if (height);
      else if (upper_rw_latch != root_leaf_rw_latch)
      {
        /* We should retry to get the page, because the root page
        is latched with different level as a leaf page. */
        ut_ad(n_blocks == 0);
        ut_ad(root_leaf_rw_latch != RW_NO_LATCH);
        upper_rw_latch= root_leaf_rw_latch;
        mtr->rollback_to_savepoint(savepoint);
        height= ULINT_UNDEFINED;
        continue;
      }
      else
      {
      reached_leaf:
        const auto leaf_savepoint= mtr->get_savepoint();
        ut_ad(leaf_savepoint);
        ut_ad(block == mtr->at_savepoint(leaf_savepoint - 1));

        if (latch_mode == BTR_MODIFY_TREE)
        {
          /* x-latch also siblings from left to right */
          if (page_has_prev(block->page.frame) &&
              !btr_latch_prev(RW_X_LATCH, block->page.id(), &err, mtr))
            break;
          if (page_has_next(block->page.frame) &&
              !btr_block_get(*index, btr_page_get_next(block->page.frame),
                             RW_X_LATCH, mtr, &err))
            break;

          if (!index->lock.have_x() &&
              btr_cur_need_opposite_intention(block->page, index->is_clust(),
                                              lock_intention,
                                              node_ptr_max_size,
                                              compress_limit, page_cur.rec))
            goto need_opposite_intention;
        }
        else
        {
          if (latch_mode != BTR_CONT_MODIFY_TREE)
          {
            ut_ad(latch_mode == BTR_MODIFY_LEAF ||
                  latch_mode == BTR_SEARCH_LEAF);
            /* Release index->lock if needed, and the non-leaf pages. */
            mtr->rollback_to_savepoint(savepoint - !latch_by_caller,
                                       leaf_savepoint - 1);
          }
        }
        break;
      }
    }
    else if (UNIV_UNLIKELY(height != l))
    {
    corrupted:
      err= DB_CORRUPTION;
      break;
    }

    if (!height)
      goto reached_leaf;

    height--;

    if (first
        ? !page_cur_move_to_next(&page_cur)
        : !page_cur_move_to_prev(&page_cur))
      goto corrupted;

    offsets= rec_get_offsets(page_cur.rec, index, offsets, 0, ULINT_UNDEFINED,
                             &heap);
    page= btr_node_ptr_get_child_page_no(page_cur.rec, offsets);

    ut_ad(latch_mode != BTR_MODIFY_TREE || upper_rw_latch == RW_X_LATCH);

    if (latch_mode != BTR_MODIFY_TREE)
    {
      if (!height && first && first_access)
        buf_read_ahead_linear(page_id_t(block->page.id().space(), page));
    }
    else if (btr_cur_need_opposite_intention(block->page, index->is_clust(),
                                             lock_intention,
                                             node_ptr_max_size, compress_limit,
                                             page_cur.rec))
    {
    need_opposite_intention:
      /* If the rec is the first or last in the page for pessimistic
      delete intention, it might cause node_ptr insert for the upper
      level. We should change the intention and retry. */

      mtr->rollback_to_savepoint(savepoint);
      mtr->index_lock_upgrade();
      /* X-latch all pages from now on */
      latch_mode= BTR_CONT_MODIFY_TREE;
      page= index->page;
      height= ULINT_UNDEFINED;
      n_blocks= 0;
      continue;
    }
    else
    {
      if (!btr_cur_will_modify_tree(index, block->page.frame,
                                    lock_intention, page_cur.rec,
                                    node_ptr_max_size,
                                    index->table->space->zip_size(), mtr))
      {
        ut_ad(n_blocks);
        /* release buffer-fixes on pages that will not be modified
        (except the root) */
        if (n_blocks > 1)
        {
          mtr->rollback_to_savepoint(savepoint + 1, savepoint + n_blocks - 1);
          n_blocks= 1;
        }
      }
    }

    /* Go to the child node */
    n_blocks++;
  }

  if (UNIV_LIKELY_NULL(heap))
    mem_heap_free(heap);

  return err;
}

/*==================== B-TREE INSERT =========================*/

/*************************************************************//**
Inserts a record if there is enough space, or if enough space can
be freed by reorganizing. Differs from btr_cur_optimistic_insert because
no heuristics is applied to whether it pays to use CPU time for
reorganizing the page or not.

@return pointer to inserted record if succeed, else NULL */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
rec_t*
btr_cur_insert_if_possible(
/*=======================*/
 btr_cur_t* cursor, /*!< in: cursor on page after which to insert;
cursor stays valid */
 const dtuple_t* tuple, /*!< in: tuple to insert; the size info need not
have been stored to tuple */
 rec_offs** offsets,/*!< out: offsets on *rec */
 mem_heap_t** heap, /*!< in/out: pointer to memory heap, or NULL */
 ulint  n_ext, /*!< in: number of externally stored columns */
 mtr_t*  mtr) /*!< in/out: mini-transaction */
{
 page_cur_t* page_cursor;
 rec_t*  rec;

 ut_ad(dtuple_check_typed(tuple));

 ut_ad(mtr->memo_contains_flagged(btr_cur_get_block(cursor),
      MTR_MEMO_PAGE_X_FIX));
 page_cursor = btr_cur_get_page_cur(cursor);

 /* Now, try the insert */
 rec = page_cur_tuple_insert(page_cursor, tuple, offsets, heap, n_ext,
        mtr);

 /* If the record did not fit, reorganize.
 For compressed pages, page_cur_tuple_insert()
attempted this already. */
 if (!rec && !page_cur_get_page_zip(page_cursor)
     && btr_page_reorganize(page_cursor, mtr) == DB_SUCCESS) {
  rec = page_cur_tuple_insert(page_cursor, tuple, offsets, heap,
         n_ext, mtr);
 }

 ut_ad(!rec || rec_offs_validate(rec, page_cursor->index, *offsets));
 return(rec);
}

/*************************************************************//**
For an insert, checks the locks and does the undo logging if desired.
@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
UNIV_INLINE MY_ATTRIBUTE((warn_unused_result, nonnull(2,3,5,6)))
dberr_t
btr_cur_ins_lock_and_undo(
/*======================*/
 ulint  flags, /*!< in: undo logging and locking flags: if
    not zero, the parameters index and thr
should be specified */
 btr_cur_t* cursor, /*!< in: cursor on page after which to insert */
 dtuple_t* entry, /*!< in/out: entry to insert */
 que_thr_t* thr, /*!< in: query thread or NULL */
 mtr_t*  mtr, /*!< in/out: mini-transaction */
 bool*  inherit)/*!< out: true if the inserted new record maybe
    should inherit LOCK_GAP type locks from the
successor record */
{
 if (!(~flags | (BTR_NO_UNDO_LOG_FLAG | BTR_KEEP_SYS_FLAG))) {
  return DB_SUCCESS;
 }

 /* Check if we have to wait for a lock: enqueue an explicit lock
request if yes */

 rec_t* rec = btr_cur_get_rec(cursor);
 dict_index_t* index = cursor->index();

 ut_ad(!dict_index_is_online_ddl(index)
       || dict_index_is_clust(index)
       || (flags & BTR_CREATE_FLAG));
 ut_ad((flags & BTR_NO_UNDO_LOG_FLAG)
       || index->table->skip_alter_undo != dict_table_t::NO_UNDO);

 ut_ad(mtr->is_named_space(index->table->space));

 /* Check if there is predicate or GAP lock preventing the insertion */
 if (!(flags & BTR_NO_LOCKING_FLAG)) {
  const unsigned type = index->type;
  if (UNIV_UNLIKELY(type & DICT_SPATIAL)) {
   lock_prdt_t prdt;
   rtr_mbr_t mbr;

   rtr_get_mbr_from_tuple(entry, &mbr);

   /* Use on stack MBR variable to test if a lock is
   needed. If so, the predicate (MBR) will be allocated
from lock heap in lock_prdt_insert_check_and_lock() */
   lock_init_prdt_from_mbr(&prdt, &mbr, 0, nullptr);

   if (dberr_t err = lock_prdt_insert_check_and_lock(
        rec, btr_cur_get_block(cursor),
        index, thr, mtr, &prdt)) {
    return err;
   }
   *inherit = false;
  } else {
   ut_ad(!dict_index_is_online_ddl(index)
         || index->is_primary()
         || (flags & BTR_CREATE_FLAG));
#ifdef WITH_WSREP
   trx_t* trx= thr_get_trx(thr);
   /* If transaction scanning an unique secondary
   key is wsrep high priority thread (brute
   force) this scanning may involve GAP-locking
   in the index. As this locking happens also
   when applying replication events in high
   priority applier threads, there is a
   probability for lock conflicts between two
   wsrep high priority threads. To avoid this
   GAP-locking we mark that this transaction
is using unique key scan here. */
   if ((type & (DICT_CLUSTERED | DICT_UNIQUE)) == DICT_UNIQUE
       && trx->is_wsrep()
       && wsrep_thd_is_BF(trx->mysql_thd, false)) {
    trx->wsrep = 3;
   }
#endif /* WITH_WSREP */
   if (dberr_t err = lock_rec_insert_check_and_lock(
        rec, btr_cur_get_block(cursor),
        index, thr, mtr, inherit)) {
    return err;
   }
  }
 }

 if (!index->is_primary() || !page_is_leaf(btr_cur_get_page(cursor))) {
  return DB_SUCCESS;
 }

 constexpr roll_ptr_t dummy_roll_ptr = roll_ptr_t{1}
  << ROLL_PTR_INSERT_FLAG_POS;
 roll_ptr_t roll_ptr = dummy_roll_ptr;

 if (!(flags & BTR_NO_UNDO_LOG_FLAG)) {
  if (dberr_t err = trx_undo_report_row_operation(
       thr, index, entry, NULL, 0, NULL, NULL,
       &roll_ptr)) {
   return err;
  }

  if (roll_ptr != dummy_roll_ptr) {
   dfield_t* r = dtuple_get_nth_field(entry,
          index->db_trx_id());
   trx_write_trx_id(static_cast<byte*>(r->data),
      thr_get_trx(thr)->id);
  }
 }

 if (!(flags & BTR_KEEP_SYS_FLAG)) {
  dfield_t* r = dtuple_get_nth_field(
   entry, index->db_roll_ptr());
  ut_ad(r->len == DATA_ROLL_PTR_LEN);
  trx_write_roll_ptr(static_cast<byte*>(r->data), roll_ptr);
 }

 return DB_SUCCESS;
}

/**
Prefetch siblings of the leaf for the pessimistic operation.
@param block leaf page
@param index    index of the page
@param trx      transaction */
static void btr_cur_prefetch_siblings(const buf_block_t *block,
                                      const dict_index_t *index,
                                      trx_t *trx) noexcept
{
  ut_ad(page_is_leaf(block->page.frame));

  const page_t *page= block->page.frame;
  uint32_t next= mach_read_from_4(my_assume_aligned<4>(page + FIL_PAGE_NEXT));
  fil_space_t *space= index->table->space;
  page_id_t id{space->id,
               mach_read_from_4(my_assume_aligned<4>(page + FIL_PAGE_PREV))};

  if (id.page_no() != FIL_NULL && space->acquire())
    buf_read_page_background(id, space, trx);

  id.set_page_no(next);

  if (next != FIL_NULL && space->acquire())
    buf_read_page_background(id, space, trx);
}

/*************************************************************//**
Tries to perform an insert to a page in an index tree, next to cursor.
It is assumed that mtr holds an x-latch on the page. The operation does
not succeed if there is too little space on the page. If there is just
one record on the page, the insert will always succeed; this is to
prevent trying to split a page with just one record.
@return DB_SUCCESS, DB_LOCK_WAIT, DB_FAIL, or error number */
dberr_t
btr_cur_optimistic_insert(
/*======================*/
 ulint  flags, /*!< in: undo logging and locking flags: if not
    zero, the parameters index and thr should be
specified */
 btr_cur_t* cursor, /*!< in: cursor on page after which to insert;
cursor stays valid */
 rec_offs** offsets,/*!< out: offsets on *rec */
 mem_heap_t** heap, /*!< in/out: pointer to memory heap */
 dtuple_t* entry, /*!< in/out: entry to insert */
 rec_t**  rec, /*!< out: pointer to inserted record if
succeed */
 big_rec_t** big_rec,/*!< out: big rec vector whose fields have to
be stored externally by the caller */
 ulint  n_ext, /*!< in: number of externally stored columns */
 que_thr_t* thr, /*!< in/out: query thread; can be NULL if
    !(~flags
    & (BTR_NO_LOCKING_FLAG
| BTR_NO_UNDO_LOG_FLAG)) */
 mtr_t*  mtr) /*!< in/out: mini-transaction;
    if this function returns DB_SUCCESS on
    a leaf page of a secondary index in a
    compressed tablespace, the caller must
    mtr_commit(mtr) before latching
any further pages */
{
 big_rec_t* big_rec_vec = NULL;
 dict_index_t* index;
 page_cur_t* page_cursor;
 buf_block_t* block;
 page_t*  page;
 rec_t*  dummy;
 bool  leaf;
 bool  reorg __attribute__((unused));
 bool  inherit = true;
 ulint  rec_size;
 dberr_t  err;

 ut_ad(thr || !(~flags & (BTR_NO_LOCKING_FLAG | BTR_NO_UNDO_LOG_FLAG)));
 *big_rec = NULL;

 block = btr_cur_get_block(cursor);
 page = buf_block_get_frame(block);
 index = cursor->index();

 ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
 ut_ad(!dict_index_is_online_ddl(index)
       || dict_index_is_clust(index)
       || (flags & BTR_CREATE_FLAG));
 ut_ad(dtuple_check_typed(entry));

#ifdef HAVE_valgrind
 if (block->page.zip.data) {
  MEM_CHECK_DEFINED(page, srv_page_size);
  MEM_CHECK_DEFINED(block->page.zip.data, block->zip_size());
 }
#endif /* HAVE_valgrind */

 leaf = page_is_leaf(page);

 if (UNIV_UNLIKELY(entry->is_alter_metadata())) {
  ut_ad(leaf);
  goto convert_big_rec;
 }

 /* Calculate the record size when entry is converted to a record */
 rec_size = rec_get_converted_size(index, entry, n_ext);

 if (page_zip_rec_needs_ext(rec_size, page_is_comp(page),
       dtuple_get_n_fields(entry),
       block->zip_size())) {
convert_big_rec:
  /* The record is so big that we have to store some fields
externally on separate database pages */
  big_rec_vec = dtuple_convert_big_rec(index, 0, entry, &n_ext);

  if (UNIV_UNLIKELY(big_rec_vec == NULL)) {

   return(DB_TOO_BIG_RECORD);
  }

  rec_size = rec_get_converted_size(index, entry, n_ext);
 }

 if (block->page.zip.data && page_zip_is_too_big(index, entry)) {
  if (big_rec_vec != NULL) {
   dtuple_convert_back_big_rec(index, entry, big_rec_vec);
  }

  return(DB_TOO_BIG_RECORD);
 }

 LIMIT_OPTIMISTIC_INSERT_DEBUG(page_get_n_recs(page), goto fail);

 if (block->page.zip.data && leaf
     && (page_get_data_size(page) + rec_size
  >= dict_index_zip_pad_optimal_page_size(index))) {
  /* If compression padding tells us that insertion will
  result in too packed up page i.e.: which is likely to
  cause compression failure then don't do an optimistic
insertion. */
fail:
  err = DB_FAIL;

  /* prefetch siblings of the leaf for the pessimistic
operation, if the page is leaf. */
  if (leaf) {
   btr_cur_prefetch_siblings(block, index, mtr->trx);
  }
fail_err:

  if (big_rec_vec) {
   dtuple_convert_back_big_rec(index, entry, big_rec_vec);
  }

  return(err);
 }

 ulint max_size = page_get_max_insert_size_after_reorganize(page, 1);
 if (max_size < rec_size) {
  goto fail;
 }

 const ulint n_recs = page_get_n_recs(page);
 if (UNIV_UNLIKELY(n_recs >= 8189)) {
  ut_ad(srv_page_size == 65536);
  goto fail;
 }

 if (page_has_garbage(page)) {
  if (max_size < BTR_CUR_PAGE_REORGANIZE_LIMIT
      && n_recs > 1
      && page_get_max_insert_size(page, 1) < rec_size) {

   goto fail;
  }
 }

 /* If there have been many consecutive inserts to the
 clustered index leaf page of an uncompressed table, check if
 we have to split the page to reserve enough free space for
future updates of records. */

 if (leaf && !block->page.zip.data && dict_index_is_clust(index)
     && page_get_n_recs(page) >= 2
     && dict_index_get_space_reserve() + rec_size > max_size
     && (btr_page_get_split_rec_to_right(cursor, &dummy)
  || btr_page_get_split_rec_to_left(cursor))) {
  goto fail;
 }

 page_cursor = btr_cur_get_page_cur(cursor);

 DBUG_LOG("ib_cur",
   "insert " << index->name << " (" << index->id << ") by "
   << ib::hex(thr ? thr->graph->trx->id : 0)
   << ' ' << rec_printer(entry).str());
 DBUG_EXECUTE_IF("do_page_reorganize",
   ut_a(!n_recs || btr_page_reorganize(page_cursor, mtr)
        == DB_SUCCESS););

 /* Now, try the insert */
 {
  const rec_t* page_cursor_rec = page_cur_get_rec(page_cursor);

  /* Check locks and write to the undo log,
if specified */
  err = btr_cur_ins_lock_and_undo(flags, cursor, entry,
      thr, mtr, &inherit);
  if (err != DB_SUCCESS) {
   goto fail_err;
  }

#ifdef UNIV_DEBUG
  if (!(flags & BTR_CREATE_FLAG)
      && leaf && index->is_primary()) {
   const dfield_t* trx_id = dtuple_get_nth_field(
    entry, dict_col_get_clust_pos(
     dict_table_get_sys_col(index->table,
              DATA_TRX_ID),
     index));

   ut_ad(trx_id->len == DATA_TRX_ID_LEN);
   ut_ad(trx_id[1].len == DATA_ROLL_PTR_LEN);
   ut_ad(*static_cast<const byte*>
         (trx_id[1].data) & 0x80);
   if (flags & BTR_NO_UNDO_LOG_FLAG) {
    ut_ad(!memcmp(trx_id->data, reset_trx_id,
           DATA_TRX_ID_LEN));
   } else {
    ut_ad(thr->graph->trx->id);
    ut_ad(thr->graph->trx->bulk_insert
          || thr->graph->trx->id
          == trx_read_trx_id(
           static_cast<const byte*>(
       trx_id->data))
          || index->table->is_temporary());
   }
  }
#endif

  *rec = page_cur_tuple_insert(page_cursor, entry, offsets, heap,
          n_ext, mtr);

  reorg = page_cursor_rec != page_cur_get_rec(page_cursor);
 }

 if (*rec) {
 } else if (block->page.zip.data) {
  ut_ad(!index->table->is_temporary());
  goto fail;
 } else {
  ut_ad(!reorg);
  reorg = true;

  /* If the record did not fit, reorganize */
  err = btr_page_reorganize(page_cursor, mtr);
  if (err != DB_SUCCESS
      || page_get_max_insert_size(page, 1) != max_size
      || !(*rec = page_cur_tuple_insert(page_cursor, entry,
            offsets, heap, n_ext,
            mtr))) {
   err = DB_CORRUPTION;
   goto fail_err;
  }
 }

#ifdef BTR_CUR_HASH_ADAPT
 if (!leaf) {
 } else if (entry->info_bits & REC_INFO_MIN_REC_FLAG) {
  ut_ad(entry->is_metadata());
  ut_ad(index->is_instant());
  ut_ad(flags == BTR_NO_LOCKING_FLAG);
 } else if (!index->table->is_temporary()) {
  btr_search_update_hash_on_insert(cursor, reorg, *mtr);
 }
#endif /* BTR_CUR_HASH_ADAPT */

 if (!(flags & BTR_NO_LOCKING_FLAG) && inherit) {

  lock_update_insert(block, *rec);
 }

 *big_rec = big_rec_vec;

 return(DB_SUCCESS);
}

/*************************************************************//**
Performs an insert on a page of an index tree. It is assumed that mtr
holds an x-latch on the tree and on the cursor page. If the insert is
made on the leaf level, to avoid deadlocks, mtr must also own x-latches
to brothers of page, if those brothers exist.
@return DB_SUCCESS or error number */
dberr_t
btr_cur_pessimistic_insert(
/*=======================*/
 ulint  flags, /*!< in: undo logging and locking flags: if not
    zero, the parameter thr should be
    specified; if no undo logging is specified,
    then the caller must have reserved enough
    free extents in the file space so that the
insertion will certainly succeed */
 btr_cur_t* cursor, /*!< in: cursor after which to insert;
cursor stays valid */
 rec_offs** offsets,/*!< out: offsets on *rec */
 mem_heap_t** heap, /*!< in/out: pointer to memory heap
that can be emptied */
 dtuple_t* entry, /*!< in/out: entry to insert */
 rec_t**  rec, /*!< out: pointer to inserted record if
succeed */
 big_rec_t** big_rec,/*!< out: big rec vector whose fields have to
be stored externally by the caller */
 ulint  n_ext, /*!< in: number of externally stored columns */
 que_thr_t* thr, /*!< in/out: query thread; can be NULL if
    !(~flags
    & (BTR_NO_LOCKING_FLAG
| BTR_NO_UNDO_LOG_FLAG)) */
 mtr_t*  mtr) /*!< in/out: mini-transaction */
{
 dict_index_t* index  = cursor->index();
 big_rec_t* big_rec_vec = NULL;
 bool  inherit = false;
 uint32_t n_reserved = 0;

 ut_d(++btr_cur_pessimistic_insert_calls);

 ut_ad(dtuple_check_typed(entry));
 ut_ad(thr || !(~flags & (BTR_NO_LOCKING_FLAG | BTR_NO_UNDO_LOG_FLAG)));

 *big_rec = NULL;

 ut_ad(mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK
      | MTR_MEMO_SX_LOCK));
 ut_ad(mtr->memo_contains_flagged(btr_cur_get_block(cursor),
      MTR_MEMO_PAGE_X_FIX));
 ut_ad(!dict_index_is_online_ddl(index)
       || dict_index_is_clust(index)
       || (flags & BTR_CREATE_FLAG));

#ifdef BTR_CUR_HASH_ADAPT
 cursor->flag = BTR_CUR_BINARY;
#endif

 /* Check locks and write to undo log, if specified */

 dberr_t err = btr_cur_ins_lock_and_undo(flags, cursor, entry,
      thr, mtr, &inherit);

 if (err != DB_SUCCESS) {
  return(err);
 }

 /* First reserve enough free space for the file segments of
 the index tree, so that the insert will not fail because of
lack of space */

 err = fsp_reserve_free_extents(&n_reserved, index->table->space,
           uint32_t(cursor->tree_height / 16 + 3),
           FSP_NORMAL, mtr);
 if (err != DB_SUCCESS) {
  return err;
 }

 if (page_zip_rec_needs_ext(rec_get_converted_size(index, entry, n_ext),
       index->table->not_redundant(),
       dtuple_get_n_fields(entry),
       btr_cur_get_block(cursor)->zip_size())
     || UNIV_UNLIKELY(entry->is_alter_metadata()
        && !dfield_is_ext(
         dtuple_get_nth_field(
          entry,
          index->first_user_field())))) {
  /* The record is so big that we have to store some fields
externally on separate database pages */

  if (UNIV_LIKELY_NULL(big_rec_vec)) {
   /* This should never happen, but we handle
the situation in a robust manner. */
   ut_ad(0);
   dtuple_convert_back_big_rec(index, entry, big_rec_vec);
  }

  big_rec_vec = dtuple_convert_big_rec(index, 0, entry, &n_ext);

  if (big_rec_vec == NULL) {

   index->table->space->release_free_extents(n_reserved);
   return(DB_TOO_BIG_RECORD);
  }
 }

 if (index->page == btr_cur_get_block(cursor)->page.id().page_no()) {
  *rec = index->is_spatial()
   ? rtr_root_raise_and_insert(flags, cursor, offsets,
          heap, entry, n_ext, mtr,
          &err, thr)
   : btr_root_raise_and_insert(flags, cursor, offsets,
          heap, entry, n_ext, mtr,
          &err);
 } else if (index->is_spatial()) {
  *rec = rtr_page_split_and_insert(flags, cursor, offsets, heap,
       entry, n_ext, mtr, &err, thr);
 } else {
  *rec = btr_page_split_and_insert(flags, cursor, offsets, heap,
       entry, n_ext, mtr, &err);
 }

 if (!*rec) {
  goto func_exit;
 }

 ut_ad(page_rec_get_next(btr_cur_get_rec(cursor)) == *rec
       || dict_index_is_spatial(index));

 if (!(flags & BTR_NO_LOCKING_FLAG)) {
  ut_ad(!index->table->is_temporary());
  if (dict_index_is_spatial(index)) {
   /* Do nothing */
  } else {
   /* The cursor might be moved to the other page
   and the max trx id field should be updated after
the cursor was fixed. */
   if (!dict_index_is_clust(index)) {
    page_update_max_trx_id(
     btr_cur_get_block(cursor),
     btr_cur_get_page_zip(cursor),
     thr_get_trx(thr)->id, mtr);
   }

   if (!page_rec_is_infimum(btr_cur_get_rec(cursor))
       || !page_has_prev(btr_cur_get_page(cursor))) {
    /* split and inserted need to call
lock_update_insert() always. */
    inherit = true;
   }
  }
 }

 if (!page_is_leaf(btr_cur_get_page(cursor))) {
  ut_ad(!big_rec_vec);
 } else {
#ifdef BTR_CUR_HASH_ADAPT
  if (entry->info_bits & REC_INFO_MIN_REC_FLAG) {
   ut_ad(entry->is_metadata());
   ut_ad(index->is_instant());
   ut_ad(flags & BTR_NO_LOCKING_FLAG);
   ut_ad(!(flags & BTR_CREATE_FLAG));
  } else if (!index->table->is_temporary()) {
   btr_search_update_hash_on_insert(cursor, false, *mtr);
  }
#endif /* BTR_CUR_HASH_ADAPT */
  if (inherit && !(flags & BTR_NO_LOCKING_FLAG)) {

   lock_update_insert(btr_cur_get_block(cursor), *rec);
  }
 }

 err = DB_SUCCESS;
func_exit:
 index->table->space->release_free_extents(n_reserved);
 *big_rec = big_rec_vec;

 return err;
}

/*==================== B-TREE UPDATE =========================*/

/*************************************************************//**
For an update, checks the locks and does the undo logging.
@return DB_SUCCESS, DB_LOCK_WAIT, or error number */
UNIV_INLINE MY_ATTRIBUTE((warn_unused_result))
dberr_t
btr_cur_upd_lock_and_undo(
/*======================*/
 ulint  flags, /*!< in: undo logging and locking flags */
 btr_cur_t* cursor, /*!< in: cursor on record to update */
 const rec_offs* offsets,/*!< in: rec_get_offsets() on cursor */
 const upd_t* update, /*!< in: update vector */
 ulint  cmpl_info,/*!< in: compiler info on secondary index
updates */
 que_thr_t* thr, /*!< in: query thread
(can be NULL if BTR_NO_LOCKING_FLAG) */
 mtr_t*  mtr, /*!< in/out: mini-transaction */
 roll_ptr_t* roll_ptr)/*!< out: roll pointer */
{
 dict_index_t* index;
 const rec_t* rec;
 dberr_t  err;

 ut_ad((thr != NULL) || (flags & BTR_NO_LOCKING_FLAG));

 rec = btr_cur_get_rec(cursor);
 index = cursor->index();

 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(mtr->is_named_space(index->table->space));

 if (!dict_index_is_clust(index)) {
  ut_ad(dict_index_is_online_ddl(index)
        == !!(flags & BTR_CREATE_FLAG));

  /* We do undo logging only when we update a clustered index
record */
  return(lock_sec_rec_modify_check_and_lock(
          flags, btr_cur_get_block(cursor), rec,
          index, thr, mtr));
 }

 /* Check if we have to wait for a lock: enqueue an explicit lock
request if yes */

 if (!(flags & BTR_NO_LOCKING_FLAG)) {
  err = lock_clust_rec_modify_check_and_lock(
   btr_cur_get_block(cursor), rec, index,
   offsets, thr);
  if (err != DB_SUCCESS) {
   return(err);
  }
 }

 /* Append the info about the update in the undo log */

 return((flags & BTR_NO_UNDO_LOG_FLAG)
        ? DB_SUCCESS
        : trx_undo_report_row_operation(
         thr, index, NULL, update,
         cmpl_info, rec, offsets, roll_ptr));
}

/** Write DB_TRX_ID,DB_ROLL_PTR to a clustered index entry.
@param[in,out] entry  clustered index entry
@param[in] index  clustered index
@param[in] trx_id  DB_TRX_ID
@param[in] roll_ptr DB_ROLL_PTR */
static void btr_cur_write_sys(
 dtuple_t*  entry,
 const dict_index_t* index,
 trx_id_t  trx_id,
 roll_ptr_t  roll_ptr)
{
 dfield_t* t = dtuple_get_nth_field(entry, index->db_trx_id());
 ut_ad(t->len == DATA_TRX_ID_LEN);
 trx_write_trx_id(static_cast<byte*>(t->data), trx_id);
 dfield_t* r = dtuple_get_nth_field(entry, index->db_roll_ptr());
 ut_ad(r->len == DATA_ROLL_PTR_LEN);
 trx_write_roll_ptr(static_cast<byte*>(r->data), roll_ptr);
}

MY_ATTRIBUTE((warn_unused_result))
/** Update DB_TRX_ID, DB_ROLL_PTR in a clustered index record.
@param[in,out]  block           clustered index leaf page
@param[in,out]  rec             clustered index record
@param[in]      index           clustered index
@param[in]      offsets         rec_get_offsets(rec, index)
@param[in]      trx             transaction
@param[in]      roll_ptr        DB_ROLL_PTR value
@param[in,out]  mtr             mini-transaction
@return error code */
static dberr_t btr_cur_upd_rec_sys(buf_block_t *block, rec_t *rec,
                                   dict_index_t *index, const rec_offs *offsets,
                                   const trx_t *trx, roll_ptr_t roll_ptr,
                                   mtr_t *mtr)
{
  ut_ad(index->is_primary());
  ut_ad(rec_offs_validate(rec, index, offsets));

  if (UNIV_LIKELY_NULL(block->page.zip.data))
  {
    page_zip_write_trx_id_and_roll_ptr(block, rec, offsets, index->db_trx_id(),
                                       trx->id, roll_ptr, mtr);
    return DB_SUCCESS;
  }

  ulint offset= index->trx_id_offset;

  if (!offset)
    offset= row_get_trx_id_offset(index, offsets);

  compile_time_assert(DATA_TRX_ID + 1 == DATA_ROLL_PTR);

  /* During IMPORT the trx id in the record can be in the future, if
  the .ibd file is being imported from another instance. During IMPORT
  roll_ptr will be 0. */
  ut_ad(roll_ptr == 0 ||
        lock_check_trx_id_sanity(trx_read_trx_id(rec + offset),
                                 rec, index, offsets));

  byte sys[DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN];

  trx_write_trx_id(sys, trx->id);
  trx_write_roll_ptr(sys + DATA_TRX_ID_LEN, roll_ptr);

  ulint d= 0;
  const byte *src= nullptr;
  byte *dest= rec + offset;
  ulint len= DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN;

  if (UNIV_LIKELY(index->trx_id_offset))
  {
    const rec_t *prev= page_rec_get_prev_const(rec);
    if (UNIV_UNLIKELY(!prev || prev == rec))
      return DB_CORRUPTION;
    else if (page_rec_is_infimum(prev));
    else
      for (src= prev + offset; d < DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN; d++)
        if (src[d] != sys[d])
          break;
    if (d > 6 && memcmp(dest, sys, d))
    {
      /* We save space by replacing a single record

      WRITE,page_offset(dest),byte[13]

      with two records:

      MEMMOVE,page_offset(dest),d(1 byte),offset(1..3 bytes),
      WRITE|0x80,0,byte[13-d]

      The single WRITE record would be x+13 bytes long, with x>2.
      The MEMMOVE record would be up to x+1+3 = x+4 bytes, and the
      second WRITE would be 1+1+13-d = 15-d bytes.

      The total size is: x+13 versus x+4+15-d = x+19-d bytes.
      To save space, we must have d>6, that is, the complete DB_TRX_ID and
      the first byte(s) of DB_ROLL_PTR must match the previous record. */
      memcpy(dest, src, d);
      mtr->memmove(*block, dest - block->page.frame, src - block->page.frame,
                   d);
      dest+= d;
      len-= d;
      /* DB_TRX_ID,DB_ROLL_PTR must be unique in each record when
      DB_TRX_ID refers to an active transaction. */
      ut_ad(len);
    }
    else
      d= 0;
  }

  if (UNIV_LIKELY(len)) /* extra safety, to avoid corrupting the log */
    mtr->memcpy<mtr_t::MAYBE_NOP>(*block, dest, sys + d, len);

  return DB_SUCCESS;
}

/*************************************************************//**
See if there is enough place in the page modification log to log
an update-in-place.

@retval false if out of space
@retval true if enough place */
bool
btr_cur_update_alloc_zip_func(
/*==========================*/
 page_zip_des_t* page_zip,/*!< in/out: compressed page */
 page_cur_t* cursor, /*!< in/out: B-tree page cursor */
#ifdef UNIV_DEBUG
 rec_offs* offsets,/*!< in/out: offsets of the cursor record */
#endif /* UNIV_DEBUG */
 ulint  length, /*!< in: size needed */
 bool  create, /*!< in: true=delete-and-insert,
false=update-in-place */
 mtr_t*  mtr) /*!< in/out: mini-transaction */
{
 dict_index_t* index = cursor->index;

 /* Have a local copy of the variables as these can change
dynamically. */
 const page_t* page = page_cur_get_page(cursor);

 ut_ad(page_zip == page_cur_get_page_zip(cursor));
 ut_ad(rec_offs_validate(page_cur_get_rec(cursor), index, offsets));

 if (page_zip_available(page_zip, dict_index_is_clust(index),
          length, create)) {
  return(true);
 }

 if (!page_zip->m_nonempty && !page_has_garbage(page)) {
  /* The page has been freshly compressed, so
reorganizing it will not help. */
  return(false);
 }

 if (create && page_is_leaf(page)
     && (length + page_get_data_size(page)
  >= dict_index_zip_pad_optimal_page_size(index))) {
  return(false);
 }

 if (btr_page_reorganize(cursor, mtr) == DB_SUCCESS) {
  rec_offs_make_valid(page_cur_get_rec(cursor), index,
        page_is_leaf(page), offsets);

  return page_zip_available(page_zip, dict_index_is_clust(index),
       length, create);
 }

 return(false);
}

/** Apply an update vector to a record. No field size changes are allowed.

This is usually invoked on a clustered index. The only use case for a
secondary index is row_ins_sec_index_entry_by_modify() or its
counterpart in ibuf_insert_to_index_page().
@param[in,out]  rec     index record
@param[in]      index   the index of the record
@param[in]      offsets rec_get_offsets(rec, index)
@param[in]      update  update vector
@param[in,out]  block   index page
@param[in,out]  mtr     mini-transaction */
void btr_cur_upd_rec_in_place(rec_t *rec, const dict_index_t *index,
                              const rec_offs *offsets, const upd_t *update,
                              buf_block_t *block, mtr_t *mtr)
{
 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(!index->table->skip_alter_undo);
 ut_ad(!block->page.zip.data || index->table->not_redundant());

#ifdef UNIV_DEBUG
 if (rec_offs_comp(offsets)) {
  switch (rec_get_status(rec)) {
  case REC_STATUS_ORDINARY:
   break;
  case REC_STATUS_INSTANT:
   ut_ad(index->is_instant());
   break;
  case REC_STATUS_NODE_PTR:
  case REC_STATUS_INFIMUM:
  case REC_STATUS_SUPREMUM:
   ut_ad("wrong record status in update" == 0);
  }
 }
#endif /* UNIV_DEBUG */

 static_assert(REC_INFO_BITS_SHIFT == 0, "compatibility");
 if (UNIV_LIKELY_NULL(block->page.zip.data)) {
  ut_ad(rec_offs_comp(offsets));
  byte* info_bits = &rec[-REC_NEW_INFO_BITS];
  const bool flip_del_mark = (*info_bits ^ update->info_bits)
   & REC_INFO_DELETED_FLAG;
  *info_bits &= byte(~REC_INFO_BITS_MASK);
  *info_bits |= update->info_bits;

  if (flip_del_mark) {
   page_zip_rec_set_deleted(block, rec, update->info_bits
       & REC_INFO_DELETED_FLAG, mtr);
  }
 } else {
  byte* info_bits = &rec[rec_offs_comp(offsets)
           ? -REC_NEW_INFO_BITS
           : -REC_OLD_INFO_BITS];

  mtr->write<1,mtr_t::MAYBE_NOP>(*block, info_bits,
            (*info_bits
      & ~REC_INFO_BITS_MASK)
            | update->info_bits);
 }

 for (ulint i = 0; i < update->n_fields; i++) {
  const upd_field_t* uf = upd_get_nth_field(update, i);
  if (upd_fld_is_virtual_col(uf) && !index->has_virtual()) {
   continue;
  }
  const ulint n = uf->field_no;

  ut_ad(!dfield_is_ext(&uf->new_val)
        == !rec_offs_nth_extern(offsets, n));
  ut_ad(!rec_offs_nth_default(offsets, n));

  if (UNIV_UNLIKELY(dfield_is_null(&uf->new_val))) {
   if (rec_offs_nth_sql_null(offsets, n)) {
    ut_ad(index->table->is_instant());
    ut_ad(n >= index->n_core_fields);
    continue;
   }

   ut_ad(!index->table->not_redundant());
   switch (ulint size = rec_get_nth_field_size(rec, n)) {
   case 0:
    break;
   case 1:
    mtr->write<1,mtr_t::MAYBE_NOP>(
     *block,
     rec_get_field_start_offs(rec, n) + rec,
     0U);
    break;
   default:
    mtr->memset(
     block,
     rec_get_field_start_offs(rec, n) + rec
     - block->page.frame,
     size, 0);
   }
   ulint l = rec_get_1byte_offs_flag(rec)
    ? (n + 1) : (n + 1) * 2;
   byte* b = rec - REC_N_OLD_EXTRA_BYTES - l;
   compile_time_assert(REC_1BYTE_SQL_NULL_MASK << 8
         == REC_2BYTE_SQL_NULL_MASK);
   mtr->write<1>(*block, b,
          byte(*b | REC_1BYTE_SQL_NULL_MASK));
   continue;
  }

  ulint len;
  byte* data = rec_get_nth_field(rec, offsets, n, &len);
  if (UNIV_LIKELY_NULL(block->page.zip.data)) {
   ut_ad(len == uf->new_val.len);
   memcpy(data, uf->new_val.data, len);
   continue;
  }

  if (UNIV_UNLIKELY(len != uf->new_val.len)) {
   ut_ad(len == UNIV_SQL_NULL);
   ut_ad(!rec_offs_comp(offsets));
   len = uf->new_val.len;
   ut_ad(len == rec_get_nth_field_size(rec, n));
   ulint l = rec_get_1byte_offs_flag(rec)
    ? (n + 1) : (n + 1) * 2;
   byte* b = rec - REC_N_OLD_EXTRA_BYTES - l;
   compile_time_assert(REC_1BYTE_SQL_NULL_MASK << 8
         == REC_2BYTE_SQL_NULL_MASK);
   mtr->write<1>(*block, b,
          byte(*b & ~REC_1BYTE_SQL_NULL_MASK));
  }

  if (len) {
   mtr->memcpy<mtr_t::MAYBE_NOP>(*block, data,
            uf->new_val.data, len);
  }
 }

 if (UNIV_LIKELY(!block->page.zip.data)) {
  return;
 }

 switch (update->n_fields) {
 case 0:
  /* We only changed the delete-mark flag. */
  return;
 case 1:
  if (!index->is_clust()
      || update->fields[0].field_no != index->db_roll_ptr()) {
   break;
  }
  goto update_sys;
 case 2:
  if (!index->is_clust()
      || update->fields[0].field_no != index->db_trx_id()
      || update->fields[1].field_no != index->db_roll_ptr()) {
   break;
  }
 update_sys:
  ulint len;
  const byte* sys = rec_get_nth_field(rec, offsets,
          index->db_trx_id(), &len);
  ut_ad(len == DATA_TRX_ID_LEN);
  page_zip_write_trx_id_and_roll_ptr(
   block, rec, offsets, index->db_trx_id(),
   trx_read_trx_id(sys),
   trx_read_roll_ptr(sys + DATA_TRX_ID_LEN), mtr);
  return;
 }

 page_zip_write_rec(block, rec, index, offsets, 0, mtr);
}

/** Check if a ROW_FORMAT=COMPRESSED page can be updated in place
@param cur     cursor pointing to ROW_FORMAT=COMPRESSED page
@param offsets rec_get_offsets(btr_cur_get_rec(cur))
@param update  index fields being updated
@param mtr     mini-transaction
@return the record in the ROW_FORMAT=COMPRESSED page
@retval nullptr if the page cannot be updated in place */
ATTRIBUTE_COLD static
rec_t *btr_cur_update_in_place_zip_check(btr_cur_t *cur, rec_offs *offsets,
                                         const upd_t& update, mtr_t *mtr)
{
  dict_index_t *index= cur->index();
  ut_ad(!index->table->is_temporary());

  switch (update.n_fields) {
  case 0:
    /* We are only changing the delete-mark flag. */
    break;
  case 1:
    if (!index->is_clust() ||
        update.fields[0].field_no != index->db_roll_ptr())
      goto check_for_overflow;
    /* We are only changing the delete-mark flag and DB_ROLL_PTR. */
    break;
  case 2:
    if (!index->is_clust() ||
        update.fields[0].field_no != index->db_trx_id() ||
        update.fields[1].field_no != index->db_roll_ptr())
      goto check_for_overflow;
    /* We are only changing DB_TRX_ID, DB_ROLL_PTR, and the delete-mark.
    They can be updated in place in the uncompressed part of the
    ROW_FORMAT=COMPRESSED page. */
    break;
  check_for_overflow:
  default:
    if (!btr_cur_update_alloc_zip(btr_cur_get_page_zip(cur),
                                  btr_cur_get_page_cur(cur),
                                  offsets, rec_offs_size(offsets),
                                  false, mtr))
      return nullptr;
  }

  return btr_cur_get_rec(cur);
}

/*************************************************************//**
Updates a record when the update causes no size changes in its fields.
We assume here that the ordering fields of the record do not change.
@return locking or undo log related error code, or
@retval DB_SUCCESS on success
@retval DB_ZIP_OVERFLOW if there is not enough space left
on a ROW_FORMAT=COMPRESSED page */
dberr_t
btr_cur_update_in_place(
/*====================*/
 ulint  flags, /*!< in: undo logging and locking flags */
 btr_cur_t* cursor, /*!< in: cursor on the record to update;
    cursor stays valid and positioned on the
same record */
 rec_offs* offsets,/*!< in/out: offsets on cursor->page_cur.rec */
 const upd_t* update, /*!< in: update vector */
 ulint  cmpl_info,/*!< in: compiler info on secondary index
updates */
 que_thr_t* thr, /*!< in: query thread */
 trx_id_t trx_id, /*!< in: transaction id */
 mtr_t*  mtr) /*!< in/out: mini-transaction; if this
    is a secondary index, the caller must
    mtr_commit(mtr) before latching any
further pages */
{
 dict_index_t* index;
 rec_t*  rec;
 roll_ptr_t roll_ptr = 0;
 ulint  was_delete_marked;

 ut_ad(page_is_leaf(cursor->page_cur.block->page.frame));
 rec = btr_cur_get_rec(cursor);
 index = cursor->index();
 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(!!page_rec_is_comp(rec) == dict_table_is_comp(index->table));
 ut_ad(trx_id > 0 || (flags & BTR_KEEP_SYS_FLAG)
       || index->table->is_temporary());
 ut_ad(dict_index_is_online_ddl(index) == !!(flags & BTR_CREATE_FLAG)
       || index->is_primary());
 ut_ad(thr_get_trx(thr)->id == trx_id
       || (flags & ulint(~BTR_KEEP_POS_FLAG))
       == (BTR_NO_UNDO_LOG_FLAG | BTR_NO_LOCKING_FLAG
    | BTR_CREATE_FLAG | BTR_KEEP_SYS_FLAG));
 ut_ad(fil_page_index_page_check(btr_cur_get_page(cursor)));
 ut_ad(btr_page_get_index_id(btr_cur_get_page(cursor)) == index->id);
 ut_ad(!(update->info_bits & REC_INFO_MIN_REC_FLAG));

 DBUG_LOG("ib_cur",
   "update-in-place " << index->name << " (" << index->id
   << ") by " << ib::hex(trx_id) << ": "
   << rec_printer(rec, offsets).str());

 buf_block_t* block = btr_cur_get_block(cursor);
 page_zip_des_t* page_zip = buf_block_get_page_zip(block);

 /* Check that enough space is available on the compressed page. */
 if (UNIV_LIKELY_NULL(page_zip)
     && !(rec = btr_cur_update_in_place_zip_check(
    cursor, offsets, *update, mtr))) {
  return DB_ZIP_OVERFLOW;
 }

 /* Do lock checking and undo logging */
 if (dberr_t err = btr_cur_upd_lock_and_undo(flags, cursor, offsets,
          update, cmpl_info,
          thr, mtr, &roll_ptr)) {
  return err;
 }

 if (flags & BTR_KEEP_SYS_FLAG) {
 } else if (dberr_t err = btr_cur_upd_rec_sys(block, rec, index, offsets,
           thr_get_trx(thr),
           roll_ptr, mtr)) {
  return err;
 }

 was_delete_marked = rec_get_deleted_flag(
  rec, page_is_comp(buf_block_get_frame(block)));
 /* In delete-marked records, DB_TRX_ID must always refer to an
existing undo log record. */
 ut_ad(!was_delete_marked
       || !dict_index_is_clust(index)
       || row_get_rec_trx_id(rec, index, offsets));

#ifdef BTR_CUR_HASH_ADAPT
 {
  auto part = block->index
   ? &btr_search.get_part(*index) : nullptr;
  if (part) {
   /* TO DO: Can we skip this if none of the fields
   index->search_info->curr_n_fields
are being updated? */

   /* The function row_upd_changes_ord_field_binary
does not work on a secondary index. */

   if (!dict_index_is_clust(index)
       || row_upd_changes_ord_field_binary(
        index, update, thr, NULL, NULL)) {
    ut_ad(!(update->info_bits
     & REC_INFO_MIN_REC_FLAG));
    /* Remove possible hash index pointer
to this record */
    btr_search_update_hash_on_delete(cursor);
   }

   part->latch.wr_lock(SRW_LOCK_CALL);
  }

  assert_block_ahi_valid(block);
#endif /* BTR_CUR_HASH_ADAPT */

  btr_cur_upd_rec_in_place(rec, index, offsets, update, block,
      mtr);

#ifdef BTR_CUR_HASH_ADAPT
  if (part) {
   part->latch.wr_unlock();
  }
 }
#endif /* BTR_CUR_HASH_ADAPT */

 if (was_delete_marked
     && !rec_get_deleted_flag(
      rec, page_is_comp(buf_block_get_frame(block)))) {
  /* The new updated record owns its possible externally
stored fields */

  btr_cur_unmark_extern_fields(block, rec, index, offsets, mtr);
 }

 return DB_SUCCESS;
}

/** Trim a metadata record during the rollback of instant ALTER TABLE.
@param[in] entry metadata tuple
@param[in] index primary key
@param[in] update update vector for the rollback
@param[in,out] trx transaction */
ATTRIBUTE_COLD
static void btr_cur_trim_alter_metadata(dtuple_t* entry,
     const dict_index_t* index,
     const upd_t* update, trx_t *trx)
{
 ut_ad(index->is_instant());
 ut_ad(update->is_alter_metadata());
 ut_ad(entry->is_alter_metadata());

 ut_ad(update->fields[0].field_no == index->first_user_field());
 ut_ad(update->fields[0].new_val.ext);
 ut_ad(update->fields[0].new_val.len == FIELD_REF_SIZE);
 ut_ad(entry->n_fields - 1 == index->n_fields);

 const byte* ptr = static_cast<const byte*>(
  update->fields[0].new_val.data);
 ut_ad(!mach_read_from_4(ptr + BTR_EXTERN_LEN));
 ut_ad(mach_read_from_4(ptr + BTR_EXTERN_LEN + 4) > 4);
 ut_ad(mach_read_from_4(ptr + BTR_EXTERN_OFFSET) == FIL_PAGE_DATA);
 ut_ad(mach_read_from_4(ptr + BTR_EXTERN_SPACE_ID)
       == index->table->space->id);

 ulint n_fields = update->fields[1].field_no;
 ut_ad(n_fields <= index->n_fields);
 if (n_fields != index->n_uniq) {
  ut_ad(n_fields
        >= index->n_core_fields);
  entry->n_fields = uint16_t(n_fields);
  return;
 }

 /* This is based on dict_table_t::deserialise_columns()
and btr_cur_instant_init_low(). */
 mtr_t mtr{trx};
 mtr.start();
 buf_block_t* block = buf_page_get(
  page_id_t(index->table->space->id,
     mach_read_from_4(ptr + BTR_EXTERN_PAGE_NO)),
  0, RW_S_LATCH, &mtr);
 if (!block) {
  ut_ad("corruption" == 0);
  mtr.commit();
  return;
 }

 btr_search_drop_page_hash_index(block, index);

 ut_ad(fil_page_get_type(block->page.frame) == FIL_PAGE_TYPE_BLOB);
 ut_ad(mach_read_from_4(&block->page.frame
          [FIL_PAGE_DATA + BTR_BLOB_HDR_NEXT_PAGE_NO])
       == FIL_NULL);
 ut_ad(mach_read_from_4(&block->page.frame
          [FIL_PAGE_DATA + BTR_BLOB_HDR_PART_LEN])
       == mach_read_from_4(ptr + BTR_EXTERN_LEN + 4));
 n_fields = mach_read_from_4(
  &block->page.frame[FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE])
  + index->first_user_field();
 /* Rollback should not increase the number of fields. */
 ut_ad(n_fields <= index->n_fields);
 ut_ad(n_fields + 1 <= entry->n_fields);
 /* dict_index_t::clear_instant_alter() cannot be invoked while
 rollback of an instant ALTER TABLE transaction is in progress
for an is_alter_metadata() record. */
 ut_ad(n_fields >= index->n_core_fields);

 mtr.commit();
 entry->n_fields = uint16_t(n_fields + 1);
}

/** Trim an update tuple due to instant ADD COLUMN, if needed.
For normal records, the trailing instantly added fields that match
the initial default values are omitted.

For the special metadata record on a table on which instant
ADD COLUMN has already been executed, both ADD COLUMN and the
rollback of ADD COLUMN need to be handled specially.

@param[in,out] entry index entry
@param[in] index index
@param[in] update update vector
@param[in] thr execution thread */
static inline
void
btr_cur_trim(
 dtuple_t*  entry,
 const dict_index_t* index,
 const upd_t*  update,
 const que_thr_t* thr)
{
 if (!index->is_instant()) {
 } else if (UNIV_UNLIKELY(update->is_metadata())) {
  /* We are either updating a metadata record
  (instant ALTER TABLE on a table where instant ALTER was
already executed) or rolling back such an operation. */
  ut_ad(!upd_get_nth_field(update, 0)->orig_len);
  ut_ad(entry->is_metadata());
  trx_t* const trx{thr->graph->trx};

  if (trx->in_rollback) {
   /* This rollback can occur either as part of
   ha_innobase::commit_inplace_alter_table() rolling
   back after a failed innobase_add_instant_try(),
   or as part of crash recovery. Either way, the
   table will be in the data dictionary cache, with
   the instantly added columns going to be removed
later in the rollback. */
   ut_ad(index->table->cached);
   /* The DB_TRX_ID,DB_ROLL_PTR are always last,
   and there should be some change to roll back.
   The first field in the update vector is the
   first instantly added column logged by
innobase_add_instant_try(). */
   ut_ad(update->n_fields > 2);
   if (update->is_alter_metadata()) {
    btr_cur_trim_alter_metadata(
     entry, index, update, trx);
    return;
   }
   ut_ad(!entry->is_alter_metadata());

   ulint n_fields = upd_get_nth_field(update, 0)
    ->field_no;
   ut_ad(n_fields + 1 >= entry->n_fields);
   entry->n_fields = uint16_t(n_fields);
  }
 } else {
  entry->trim(*index);
 }
}

/*************************************************************//**
Tries to update a record on a page in an index tree. It is assumed that mtr
holds an x-latch on the page. The operation does not succeed if there is too
little space on the page or if the update would result in too empty a page,
so that tree compression is recommended. We assume here that the ordering
fields of the record do not change.
@return error code, including
@retval DB_SUCCESS on success
@retval DB_OVERFLOW if the updated record does not fit
@retval DB_UNDERFLOW if the page would become too empty
@retval DB_ZIP_OVERFLOW if there is not enough space left
on a ROW_FORMAT=COMPRESSED page */
dberr_t
btr_cur_optimistic_update(
/*======================*/
 ulint  flags, /*!< in: undo logging and locking flags */
 btr_cur_t* cursor, /*!< in: cursor on the record to update;
    cursor stays valid and positioned on the
same record */
 rec_offs** offsets,/*!< out: offsets on cursor->page_cur.rec */
 mem_heap_t** heap, /*!< in/out: pointer to NULL or memory heap */
 const upd_t* update, /*!< in: update vector; this must also
contain trx id and roll ptr fields */
 ulint  cmpl_info,/*!< in: compiler info on secondary index
updates */
 que_thr_t* thr, /*!< in: query thread */
 trx_id_t trx_id, /*!< in: transaction id */
 mtr_t*  mtr) /*!< in/out: mini-transaction; if this
    is a secondary index, the caller must
    mtr_commit(mtr) before latching any
further pages */
{
 dict_index_t* index;
 page_cur_t* page_cursor;
 dberr_t  err;
 buf_block_t* block;
 page_t*  page;
 page_zip_des_t* page_zip;
 rec_t*  rec;
 ulint  max_size;
 ulint  new_rec_size;
 ulint  old_rec_size;
 dtuple_t* new_entry;
 roll_ptr_t roll_ptr;
 ulint  i;

 block = btr_cur_get_block(cursor);
 page = buf_block_get_frame(block);
 rec = btr_cur_get_rec(cursor);
 index = cursor->index();
 ut_ad(trx_id > 0 || (flags & BTR_KEEP_SYS_FLAG)
       || index->table->is_temporary());
 ut_ad(!!page_rec_is_comp(rec) == dict_table_is_comp(index->table));
 ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
 /* This is intended only for leaf page updates */
 ut_ad(page_is_leaf(page));
 ut_ad(dict_index_is_online_ddl(index) == !!(flags & BTR_CREATE_FLAG)
       || dict_index_is_clust(index));
 ut_ad(thr_get_trx(thr)->id == trx_id
       || (flags & ulint(~BTR_KEEP_POS_FLAG))
       == (BTR_NO_UNDO_LOG_FLAG | BTR_NO_LOCKING_FLAG
    | BTR_CREATE_FLAG | BTR_KEEP_SYS_FLAG));
 ut_ad(fil_page_index_page_check(page));
 ut_ad(btr_page_get_index_id(page) == index->id);

 *offsets = rec_get_offsets(rec, index, *offsets, index->n_core_fields,
       ULINT_UNDEFINED, heap);
#if defined UNIV_DEBUG || defined UNIV_BLOB_LIGHT_DEBUG
 /* Blob pointer can be null if InnoDB was killed or
ran out of space while allocating a page. */
 ut_a(!rec_offs_any_null_extern(rec, *offsets)
      || thr_get_trx(thr)->in_rollback);
#endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */

 if (UNIV_LIKELY(!update->is_metadata())
     && !row_upd_changes_field_size_or_external(index, *offsets,
             update)) {

  /* The simplest and the most common case: the update does not
  change the size of any field and none of the updated fields is
  externally stored in rec or update, and there is enough space
on the compressed page to log the update. */

  return(btr_cur_update_in_place(
          flags, cursor, *offsets, update,
          cmpl_info, thr, trx_id, mtr));
 }

 if (rec_offs_any_extern(*offsets)) {
any_extern:
  /* Externally stored fields are treated in pessimistic
update */

  /* prefetch siblings of the leaf for the pessimistic
operation. */
  btr_cur_prefetch_siblings(block, index, mtr->trx);

  return(DB_OVERFLOW);
 }

 if (rec_is_metadata(rec, *index) && index->table->instant) {
  goto any_extern;
 }

 for (i = 0; i < upd_get_n_fields(update); i++) {
  if (dfield_is_ext(&upd_get_nth_field(update, i)->new_val)) {

   goto any_extern;
  }
 }

 DBUG_LOG("ib_cur",
   "update " << index->name << " (" << index->id << ") by "
   << ib::hex(trx_id) << ": "
   << rec_printer(rec, *offsets).str());

 page_cursor = btr_cur_get_page_cur(cursor);

 if (!*heap) {
  *heap = mem_heap_create(
   rec_offs_size(*offsets)
   + DTUPLE_EST_ALLOC(rec_offs_n_fields(*offsets)));
 }

 new_entry = row_rec_to_index_entry(rec, index, *offsets, *heap);
 ut_ad(!dtuple_get_n_ext(new_entry));

 /* The page containing the clustered index record
 corresponding to new_entry is latched in mtr.
Thus the following call is safe. */
 row_upd_index_replace_new_col_vals_index_pos(new_entry, index, update,
           *heap);
 btr_cur_trim(new_entry, index, update, thr);
 old_rec_size = rec_offs_size(*offsets);
 new_rec_size = rec_get_converted_size(index, new_entry, 0);

 page_zip = buf_block_get_page_zip(block);
#ifdef UNIV_ZIP_DEBUG
 ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */

 if (page_zip) {
  ut_ad(!index->table->is_temporary());

  if (page_zip_rec_needs_ext(new_rec_size, page_is_comp(page),
        dict_index_get_n_fields(index),
        block->zip_size())) {
   goto any_extern;
  }

  if (!btr_cur_update_alloc_zip(
       page_zip, page_cursor, *offsets,
       new_rec_size, true, mtr)) {
   return(DB_ZIP_OVERFLOW);
  }

  rec = page_cur_get_rec(page_cursor);
 }

 /* We limit max record size to 16k even for 64k page size. */
 if (new_rec_size >= COMPRESSED_REC_MAX_DATA_SIZE ||
   (!dict_table_is_comp(index->table)
    && new_rec_size >= REDUNDANT_REC_MAX_DATA_SIZE)) {
  err = DB_OVERFLOW;
  goto func_exit;
 }

 if (UNIV_UNLIKELY(new_rec_size
     >= (page_get_free_space_of_empty(page_is_comp(page))
         / 2))) {
  err = DB_OVERFLOW;
  goto func_exit;
 }

 if (UNIV_UNLIKELY(page_get_data_size(page)
     - old_rec_size + new_rec_size
     < BTR_CUR_PAGE_COMPRESS_LIMIT(index))) {
  /* The page would become too empty. When innodb_index_shrink
  is OFF, only treat this as DB_UNDERFLOW if the record is
  actually shrinking; otherwise a freshly split page would be
re-merged even though the update is growing the record. */
  if (new_rec_size < old_rec_size
      || btr_cur_index_shrink) {
   err = DB_UNDERFLOW;
   goto func_exit;
  }
 }

 /* We do not attempt to reorganize if the page is compressed.
This is because the page may fail to compress after reorganization. */
 max_size = page_zip
  ? page_get_max_insert_size(page, 1)
  : (old_rec_size
     + page_get_max_insert_size_after_reorganize(page, 1));

 if (!(((max_size >= BTR_CUR_PAGE_REORGANIZE_LIMIT)
        && (max_size >= new_rec_size))
       || (page_get_n_recs(page) <= 1))) {
  /* There was not enough space, or it did not pay to
  reorganize: for simplicity, we decide what to do assuming a
reorganization is needed, though it might not be necessary */

  err = DB_OVERFLOW;
  goto func_exit;
 }

 /* Do lock checking and undo logging */
 err = btr_cur_upd_lock_and_undo(flags, cursor, *offsets,
     update, cmpl_info,
     thr, mtr, &roll_ptr);
 if (err != DB_SUCCESS) {
  goto func_exit;
 }

 /* Ok, we may do the replacement. Store on the page infimum the
 explicit locks on rec, before deleting rec (see the comment in
btr_cur_pessimistic_update). */
 if (index->has_locking()) {
  lock_rec_store_on_page_infimum(block, rec);
 }

 if (UNIV_UNLIKELY(update->is_metadata())) {
  ut_ad(new_entry->is_metadata());
  ut_ad(index->is_instant());
  /* This can be innobase_add_instant_try() performing a
  subsequent instant ADD COLUMN, or its rollback by
row_undo_mod_clust_low(). */
  ut_ad(flags & BTR_NO_LOCKING_FLAG);
 } else {
  btr_search_update_hash_on_delete(cursor);
 }

 page_cur_delete_rec(page_cursor, *offsets, mtr);

 if (!page_cur_move_to_prev(page_cursor)) {
  return DB_CORRUPTION;
 }

 if (!(flags & BTR_KEEP_SYS_FLAG)) {
  btr_cur_write_sys(new_entry, index, trx_id, roll_ptr);
 }

 rec = btr_cur_insert_if_possible(cursor, new_entry, offsets, heap,
      0/*n_ext*/, mtr);
 if (UNIV_UNLIKELY(!rec)) {
  goto corrupted;
 }

 if (UNIV_UNLIKELY(update->is_metadata())) {
  /* We must empty the PAGE_FREE list, because if this
  was a rollback, the shortened metadata record
  would have too many fields, and we would be unable to
know the size of the freed record. */
  err = btr_page_reorganize(page_cursor, mtr);
  if (err != DB_SUCCESS) {
   goto func_exit;
  }
 } else {
  /* Restore the old explicit lock state on the record */
  lock_rec_restore_from_page_infimum(*block, rec,
         block->page.id());
 }

 ut_ad(err == DB_SUCCESS);
 if (!page_cur_move_to_next(page_cursor)) {
corrupted:
  return DB_CORRUPTION;
 }

 if (err != DB_SUCCESS) {
func_exit:
  /* prefetch siblings of the leaf for the pessimistic
operation. */
  btr_cur_prefetch_siblings(block, index, mtr->trx);
 }

 return(err);
}

/*************************************************************//**
If, in a split, a new supremum record was created as the predecessor of the
updated record, the supremum record must inherit exactly the locks on the
updated record. In the split it may have inherited locks from the successor
of the updated record, which is not correct. This function restores the
right locks for the new supremum. */
static
dberr_t
btr_cur_pess_upd_restore_supremum(
/*==============================*/
 buf_block_t* block, /*!< in: buffer block of rec */
 const rec_t* rec, /*!< in: updated record */
 mtr_t*  mtr) /*!< in: mtr */
{
 page_t*  page;

 page = buf_block_get_frame(block);

 if (page_rec_get_next(page_get_infimum_rec(page)) != rec) {
  /* Updated record is not the first user record on its page */
  return DB_SUCCESS;
 }

 const uint32_t prev_page_no = btr_page_get_prev(page);

 const page_id_t block_id{block->page.id()};
 const page_id_t prev_id(block_id.space(), prev_page_no);
 buf_block_t* prev_block
  = mtr->get_already_latched(prev_id, MTR_MEMO_PAGE_X_FIX);
 if (UNIV_UNLIKELY(!prev_block)) {
  return DB_CORRUPTION;
 }
 ut_ad(!memcmp_aligned<4>(prev_block->page.frame + FIL_PAGE_NEXT,
     block->page.frame + FIL_PAGE_OFFSET, 4));

 lock_rec_reset_and_inherit_gap_locks(*prev_block, block_id,
          PAGE_HEAP_NO_SUPREMUM,
          page_is_comp(page)
          ? rec_get_heap_no_new(rec)
          : rec_get_heap_no_old(rec));
 return DB_SUCCESS;
}

/*************************************************************//**
Performs an update of a record on a page of a tree. It is assumed
that mtr holds an x-latch on the tree and on the cursor page. If the
update is made on the leaf level, to avoid deadlocks, mtr must also
own x-latches to brothers of page, if those brothers exist. We assume
here that the ordering fields of the record do not change.
@return DB_SUCCESS or error code */
dberr_t
btr_cur_pessimistic_update(
/*=======================*/
 ulint  flags, /*!< in: undo logging, locking, and rollback
flags */
 btr_cur_t* cursor, /*!< in/out: cursor on the record to update;
    cursor may become invalid if *big_rec == NULL
|| !(flags & BTR_KEEP_POS_FLAG) */
 rec_offs** offsets,/*!< out: offsets on cursor->page_cur.rec */
 mem_heap_t** offsets_heap,
    /*!< in/out: pointer to memory heap
that can be emptied */
 mem_heap_t* entry_heap,
    /*!< in/out: memory heap for allocating
big_rec and the index tuple */
 big_rec_t** big_rec,/*!< out: big rec vector whose fields have to
be stored externally by the caller */
 upd_t*  update, /*!< in/out: update vector; this is allowed to
    also contain trx id and roll ptr fields.
    Non-updated columns that are moved offpage will
be appended to this. */
 ulint  cmpl_info,/*!< in: compiler info on secondary index
updates */
 que_thr_t* thr, /*!< in: query thread */
 trx_id_t trx_id, /*!< in: transaction id */
 mtr_t*  mtr) /*!< in/out: mini-transaction; must be
committed before latching any further pages */
{
 big_rec_t* big_rec_vec = NULL;
 big_rec_t* dummy_big_rec;
 dict_index_t* index;
 buf_block_t* block;
 rec_t*  rec;
 page_cur_t* page_cursor;
 dberr_t  err;
 dberr_t  optim_err;
 roll_ptr_t roll_ptr;
 bool  was_first;
 uint32_t n_reserved = 0;

 *offsets = NULL;
 *big_rec = NULL;

 block = btr_cur_get_block(cursor);
 index = cursor->index();

 ut_d(++btr_cur_pessimistic_update_calls);

 ut_ad(mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK |
      MTR_MEMO_SX_LOCK));
 ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
#if defined UNIV_ZIP_DEBUG || defined UNIV_DEBUG
 page_zip_des_t* page_zip = buf_block_get_page_zip(block);
#endif
#ifdef UNIV_ZIP_DEBUG
 ut_a(!page_zip
      || page_zip_validate(page_zip, block->page.frame, index));
#endif /* UNIV_ZIP_DEBUG */
 ut_ad(!page_zip || !index->table->is_temporary());
 ut_ad(trx_id > 0 || (flags & BTR_KEEP_SYS_FLAG)
       || index->table->is_temporary());
 ut_ad(dict_index_is_online_ddl(index) == !!(flags & BTR_CREATE_FLAG)
       || dict_index_is_clust(index));
 ut_ad(thr_get_trx(thr)->id == trx_id
       || (flags & ulint(~BTR_KEEP_POS_FLAG))
       == (BTR_NO_UNDO_LOG_FLAG | BTR_NO_LOCKING_FLAG
    | BTR_CREATE_FLAG | BTR_KEEP_SYS_FLAG));

 err = optim_err = btr_cur_optimistic_update(
  flags,
  cursor, offsets, offsets_heap, update,
  cmpl_info, thr, trx_id, mtr);

 switch (err) {
 case DB_ZIP_OVERFLOW:
  break;
 case DB_UNDERFLOW:
  ut_d(++btr_cur_pessimistic_update_optim_err_underflows);
  break;
 case DB_OVERFLOW:
  ut_d(++btr_cur_pessimistic_update_optim_err_overflows);
  break;
 default:
 err_exit:
  if (big_rec_vec != NULL) {
   dtuple_big_rec_free(big_rec_vec);
  }

  return(err);
 }

 rec = btr_cur_get_rec(cursor);
 ut_ad(rec_offs_validate(rec, index, *offsets));

 dtuple_t* new_entry;

 const bool is_metadata = rec_is_metadata(rec, *index);

 if (UNIV_UNLIKELY(is_metadata)) {
  ut_ad(update->is_metadata());
  ut_ad(flags & BTR_NO_LOCKING_FLAG);
  ut_ad(index->is_instant());
  new_entry = row_metadata_to_tuple(
   rec, index, *offsets, entry_heap,
   update->info_bits, !thr_get_trx(thr)->in_rollback);
  ut_ad(new_entry->n_fields
        == ulint(index->n_fields)
        + update->is_alter_metadata());
 } else {
  new_entry = row_rec_to_index_entry(rec, index, *offsets,
         entry_heap);
 }

 /* The page containing the clustered index record
 corresponding to new_entry is latched in mtr.  If the
 clustered index record is delete-marked, then its externally
 stored fields cannot have been purged yet, because then the
 purge would also have removed the clustered index record
itself.  Thus the following call is safe. */
 row_upd_index_replace_new_col_vals_index_pos(new_entry, index, update,
           entry_heap);
 btr_cur_trim(new_entry, index, update, thr);

 /* We have to set appropriate extern storage bits in the new
record to be inserted: we have to remember which fields were such */

 ut_ad(!page_is_comp(block->page.frame) || !rec_get_node_ptr_flag(rec));
 ut_ad(rec_offs_validate(rec, index, *offsets));

 if ((flags & BTR_NO_UNDO_LOG_FLAG)
     && rec_offs_any_extern(*offsets)) {
  /* We are in a transaction rollback undoing a row
  update: we must free possible externally stored fields
  which got new values in the update, if they are not
  inherited values. They can be inherited if we have
  updated the primary key to another value, and then
update it back again. */

  ut_ad(big_rec_vec == NULL);
  ut_ad(dict_index_is_clust(index));
  ut_ad(thr_get_trx(thr)->in_rollback);

  DEBUG_SYNC_C("blob_rollback_middle");

  btr_rec_free_updated_extern_fields(
   index, rec, block, *offsets, update, true, mtr);
 }

 ulint n_ext = index->is_primary() ? dtuple_get_n_ext(new_entry) : 0;

 if (page_zip_rec_needs_ext(
      rec_get_converted_size(index, new_entry, n_ext),
      page_is_comp(block->page.frame),
      dict_index_get_n_fields(index),
      block->zip_size())
     || (UNIV_UNLIKELY(update->is_alter_metadata())
  && !dfield_is_ext(dtuple_get_nth_field(
       new_entry,
       index->first_user_field())))) {
  big_rec_vec = dtuple_convert_big_rec(index, update, new_entry, &n_ext);
  if (UNIV_UNLIKELY(big_rec_vec == NULL)) {
#ifdef UNIV_ZIP_DEBUG
   ut_a(!page_zip
        || page_zip_validate(page_zip, block->page.frame,
        index));
#endif /* UNIV_ZIP_DEBUG */
   index->table->space->release_free_extents(n_reserved);
   err = DB_TOO_BIG_RECORD;
   goto err_exit;
  }

  ut_ad(page_is_leaf(block->page.frame));
  ut_ad(dict_index_is_clust(index));
  if (UNIV_UNLIKELY(!(flags & BTR_KEEP_POS_FLAG))) {
   ut_ad(page_zip != NULL);
   dtuple_convert_back_big_rec(index, new_entry,
          big_rec_vec);
   big_rec_vec = NULL;
   n_ext = dtuple_get_n_ext(new_entry);
  }
 }

 /* Do lock checking and undo logging */
 err = btr_cur_upd_lock_and_undo(flags, cursor, *offsets,
     update, cmpl_info,
     thr, mtr, &roll_ptr);
 if (err != DB_SUCCESS) {
  goto err_exit;
 }

 if (optim_err == DB_OVERFLOW) {
  /* First reserve enough free space for the file segments
  of the index tree, so that the update will not fail because
of lack of space */

  err = fsp_reserve_free_extents(
   &n_reserved, index->table->space,
   uint32_t(cursor->tree_height / 16 + 3),
   flags & BTR_NO_UNDO_LOG_FLAG
   ? FSP_CLEANING : FSP_NORMAL,
   mtr);
                if (UNIV_UNLIKELY(err != DB_SUCCESS)) {
   err = DB_OUT_OF_FILE_SPACE;
   goto err_exit;
  }
 }

 if (!(flags & BTR_KEEP_SYS_FLAG)) {
  btr_cur_write_sys(new_entry, index, trx_id, roll_ptr);
 }

 if (UNIV_UNLIKELY(is_metadata)) {
  ut_ad(new_entry->is_metadata());
  ut_ad(index->is_instant());
  /* This can be innobase_add_instant_try() performing a
  subsequent instant ALTER TABLE, or its rollback by
row_undo_mod_clust_low(). */
  ut_ad(flags & BTR_NO_LOCKING_FLAG);
 } else {
  btr_search_update_hash_on_delete(cursor);

  /* Store state of explicit locks on rec on the page
  infimum record, before deleting rec. The page infimum
  acts as a dummy carrier of the locks, taking care also
  of lock releases, before we can move the locks back on
  the actual record. There is a special case: if we are
  inserting on the root page and the insert causes a
  call of btr_root_raise_and_insert. Therefore we cannot
  in the lock system delete the lock structs set on the
  root page even if the root page carries just node
pointers. */
  lock_rec_store_on_page_infimum(block, rec);
 }

#ifdef UNIV_ZIP_DEBUG
 ut_a(!page_zip
      || page_zip_validate(page_zip, block->page.frame, index));
#endif /* UNIV_ZIP_DEBUG */
 page_cursor = btr_cur_get_page_cur(cursor);

 page_cur_delete_rec(page_cursor, *offsets, mtr);

 if (!page_cur_move_to_prev(page_cursor)) {
  err = DB_CORRUPTION;
  goto return_after_reservations;
 }

 /* Force a page split instead of the in-place reinsert below when
 (cheapest checks first; the most expensive one last):
   - optimistic update returned DB_OVERFLOW (the fit problem this
     optimization addresses; DB_UNDERFLOW means the page is too sparse
     and the legacy compress path is the right answer);
   - innodb_index_shrink is OFF (the opt-in);
   - uncompressed page: the reorganize-fit check uses uncompressed-page
     accounting and is not meaningful for ROW_FORMAT=COMPRESSED;
   - page still has at least one record after the delete above (so
     the forced split would produce at least 1+1, not 0+1);
   - the page is too full to satisfy BTR_CUR_PAGE_REORGANIZE_LIMIT
     after a reorganize, so the legacy in-place retry would just churn;
   - the new record is strictly larger than the old one (only
     size-growing UPDATEs should pay the cost of an early split). This
     rec_get_converted_size() walks every field, so it is checked last.
rec=NULL falls through to the split path. */
 if (optim_err == DB_OVERFLOW
     && !btr_cur_index_shrink
     && !buf_block_get_page_zip(block)
     && page_get_n_recs(block->page.frame) > 0
     && page_get_max_insert_size_after_reorganize(
         block->page.frame, 1) < BTR_CUR_PAGE_REORGANIZE_LIMIT
     && rec_get_converted_size(index, new_entry, n_ext)
        > rec_offs_size(*offsets)) {
  rec = NULL;
 } else {
  rec = btr_cur_insert_if_possible(cursor, new_entry,
       offsets, offsets_heap, n_ext, mtr);
 }

 if (rec) {
  page_cursor->rec = rec;

  if (UNIV_UNLIKELY(is_metadata)) {
   /* We must empty the PAGE_FREE list, because if this
   was a rollback, the shortened metadata record
   would have too many fields, and we would be unable to
know the size of the freed record. */
   err = btr_page_reorganize(page_cursor, mtr);
   if (err != DB_SUCCESS) {
    goto return_after_reservations;
   }
   rec = page_cursor->rec;
   rec_offs_make_valid(rec, index, true, *offsets);
   if (page_cursor->block->page.id().page_no()
       == index->page) {
    btr_set_instant(page_cursor->block, *index,
      mtr);
   }
  } else {
   lock_rec_restore_from_page_infimum(
    *btr_cur_get_block(cursor), rec,
    block->page.id());
  }

  if (!rec_get_deleted_flag(rec, rec_offs_comp(*offsets))
      || rec_is_alter_metadata(rec, *index)) {
   /* The new inserted record owns its possible externally
stored fields */
   btr_cur_unmark_extern_fields(btr_cur_get_block(cursor),
           rec, index, *offsets, mtr);
  } else {
   /* In delete-marked records, DB_TRX_ID must
always refer to an existing undo log record. */
   ut_ad(row_get_rec_trx_id(rec, index, *offsets));
  }

  bool adjust = big_rec_vec && (flags & BTR_KEEP_POS_FLAG);
  ut_ad(!adjust || page_is_leaf(block->page.frame));

  if (btr_cur_compress_if_useful(cursor, adjust, mtr)) {
   if (adjust) {
    rec_offs_make_valid(page_cursor->rec, index,
          true, *offsets);
   }
  }

#if 0 // FIXME: this used to be a no-op, and will cause trouble if enabled
  if (!big_rec_vec
      && page_is_leaf(block->page.frame)
      && !dict_index_is_online_ddl(index)) {
   mtr->release(index->lock);
   /* NOTE: We cannot release root block latch here, because it
has segment header and already modified in most of cases.*/
  }
#endif

  err = DB_SUCCESS;
  goto return_after_reservations;
 } else {
  /* If the page is compressed and it initially
  compresses very well, and there is a subsequent insert
  of a badly-compressing record, it is possible for
  btr_cur_optimistic_update() to return DB_UNDERFLOW and
btr_cur_insert_if_possible() to return FALSE. */
  ut_ad(page_zip || optim_err != DB_UNDERFLOW);
 }

 if (big_rec_vec != NULL) {
  ut_ad(page_is_leaf(block->page.frame));
  ut_ad(dict_index_is_clust(index));
  ut_ad(flags & BTR_KEEP_POS_FLAG);

  /* btr_page_split_and_insert() in
  btr_cur_pessimistic_insert() invokes
  mtr->release(index->lock).
  We must keep the index->lock when we created a
  big_rec, so that row_upd_clust_rec() can store the
big_rec in the same mini-transaction. */

  ut_ad(mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK
       | MTR_MEMO_SX_LOCK));
  mtr_sx_lock_index(index, mtr);
 }

 /* Was the record to be updated positioned as the first user
record on its page? */
 was_first = page_cur_is_before_first(page_cursor);

 /* Lock checks and undo logging were already performed by
 btr_cur_upd_lock_and_undo(). We do not try
 btr_cur_optimistic_insert() because
btr_cur_insert_if_possible() already failed above. */

 err = btr_cur_pessimistic_insert(BTR_NO_UNDO_LOG_FLAG
      | BTR_NO_LOCKING_FLAG
      | BTR_KEEP_SYS_FLAG,
      cursor, offsets, offsets_heap,
      new_entry, &rec,
      &dummy_big_rec, n_ext, NULL, mtr);
 if (err) {
  /* This should happen when InnoDB tries to extend the
tablespace */
  ut_ad(err == DB_OUT_OF_FILE_SPACE);
  return err;
 }
 ut_a(rec);
 ut_a(dummy_big_rec == NULL);
 ut_ad(rec_offs_validate(rec, cursor->index(), *offsets));
 page_cursor->rec = rec;

 /* Multiple transactions cannot simultaneously operate on the
 same temp-table in parallel.
 max_trx_id is ignored for temp tables because it not required
for MVCC. */
 if (!index->is_primary() && !index->table->is_temporary()) {
  /* Update PAGE_MAX_TRX_ID in the index page header.
  It was not updated by btr_cur_pessimistic_insert()
because of BTR_NO_LOCKING_FLAG. */
  page_update_max_trx_id(btr_cur_get_block(cursor),
           btr_cur_get_page_zip(cursor),
           trx_id, mtr);
 }

 if (!rec_get_deleted_flag(rec, rec_offs_comp(*offsets))) {
  /* The new inserted record owns its possible externally
stored fields */
#ifdef UNIV_ZIP_DEBUG
  ut_a(!page_zip
       || page_zip_validate(page_zip, block->page.frame, index));
#endif /* UNIV_ZIP_DEBUG */
  btr_cur_unmark_extern_fields(btr_cur_get_block(cursor), rec,
          index, *offsets, mtr);
 } else {
  /* In delete-marked records, DB_TRX_ID must
always refer to an existing undo log record. */
  ut_ad(row_get_rec_trx_id(rec, index, *offsets));
 }

 if (UNIV_UNLIKELY(is_metadata)) {
  /* We must empty the PAGE_FREE list, because if this
  was a rollback, the shortened metadata record
  would have too many fields, and we would be unable to
know the size of the freed record. */
  err = btr_page_reorganize(page_cursor, mtr);
  if (err != DB_SUCCESS) {
   goto return_after_reservations;
  }
  rec = page_cursor->rec;
 } else {
  lock_rec_restore_from_page_infimum(
   *btr_cur_get_block(cursor), rec, block->page.id());
 }

 /* If necessary, restore also the correct lock state for a new,
 preceding supremum record created in a page split. While the old
 record was nonexistent, the supremum might have inherited its locks
from a wrong record. */

 if (!was_first) {
  err = btr_cur_pess_upd_restore_supremum(
   btr_cur_get_block(cursor), rec, mtr);
 }

return_after_reservations:
#ifdef UNIV_ZIP_DEBUG
 ut_a(err ||
      !page_zip || page_zip_validate(btr_cur_get_page_zip(cursor),
         btr_cur_get_page(cursor), index));
#endif /* UNIV_ZIP_DEBUG */

 index->table->space->release_free_extents(n_reserved);
 *big_rec = big_rec_vec;
 return(err);
}

/*==================== B-TREE DELETE MARK AND UNMARK ===============*/

/** Modify the delete-mark flag of a record.
@tparam         flag    the value of the delete-mark flag
@param[in,out]  block   buffer block
@param[in,out]  rec     record on a physical index page
@param[in,out]  mtr     mini-transaction  */
template<bool flag>
void btr_rec_set_deleted(buf_block_t *block, rec_t *rec, mtr_t *mtr)
{
  if (UNIV_LIKELY(page_is_comp(block->page.frame) != 0))
  {
    byte *b= &rec[-REC_NEW_INFO_BITS];
    const byte v= flag
      ? (*b | REC_INFO_DELETED_FLAG)
      : (*b & byte(~REC_INFO_DELETED_FLAG));
    if (*b == v);
    else if (UNIV_LIKELY_NULL(block->page.zip.data))
    {
      *b= v;
      page_zip_rec_set_deleted(block, rec, flag, mtr);
    }
    else
      mtr->write<1>(*block, b, v);
  }
  else
  {
    ut_ad(!block->page.zip.data);
    byte *b= &rec[-REC_OLD_INFO_BITS];
    const byte v = flag
      ? (*b | REC_INFO_DELETED_FLAG)
      : (*b & byte(~REC_INFO_DELETED_FLAG));
    mtr->write<1,mtr_t::MAYBE_NOP>(*block, b, v);
  }
}

template void btr_rec_set_deleted<false>(buf_block_t *, rec_t *, mtr_t *);
template void btr_rec_set_deleted<true>(buf_block_t *, rec_t *, mtr_t *);

/***********************************************************//**
Marks a clustered index record deleted. Writes an undo log record to
undo log on this delete marking. Writes in the trx id field the id
of the deleting transaction, and in the roll ptr field pointer to the
undo log record created.
@return DB_SUCCESS, DB_LOCK_WAIT, or error number */
dberr_t
btr_cur_del_mark_set_clust_rec(
/*===========================*/
 buf_block_t* block, /*!< in/out: buffer block of the record */
 rec_t*  rec, /*!< in/out: record */
 dict_index_t* index, /*!< in: clustered index of the record */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec) */
 que_thr_t* thr, /*!< in: query thread */
 const dtuple_t* entry, /*!< in: dtuple for the deleting record, also
contains the virtual cols if there are any */
 mtr_t*  mtr) /*!< in/out: mini-transaction */
{
 roll_ptr_t roll_ptr;
 dberr_t  err;
 trx_t*  trx;

 ut_ad(dict_index_is_clust(index));
 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(!!page_rec_is_comp(rec) == dict_table_is_comp(index->table));
 ut_ad(buf_block_get_frame(block) == page_align(rec));
 ut_ad(page_rec_is_leaf(rec));
 ut_ad(mtr->is_named_space(index->table->space));

 if (rec_get_deleted_flag(rec, rec_offs_comp(offsets))) {
  /* We may already have delete-marked this record
when executing an ON DELETE CASCADE operation. */
  ut_ad(row_get_rec_trx_id(rec, index, offsets)
        == thr_get_trx(thr)->id);
  return(DB_SUCCESS);
 }

 err = trx_undo_report_row_operation(thr, index,
         entry, NULL, 0, rec, offsets,
         &roll_ptr);
 if (err != DB_SUCCESS) {

  return(err);
 }

 /* The search latch is not needed here, because
 the adaptive hash index does not depend on the delete-mark
and the delete-mark is being updated in place. */

 btr_rec_set_deleted<true>(block, rec, mtr);

 trx = thr_get_trx(thr);

 DBUG_LOG("ib_cur",
   "delete-mark clust " << index->table->name
   << " (" << index->id << ") by "
   << ib::hex(trx->id) << ": "
   << rec_printer(rec, offsets).str());

 return btr_cur_upd_rec_sys(block, rec, index, offsets, trx, roll_ptr,
       mtr);
}

/*==================== B-TREE RECORD REMOVE =========================*/

/*************************************************************//**
Tries to compress a page of the tree if it seems useful. It is assumed
that mtr holds an x-latch on the tree and on the cursor page. To avoid
deadlocks, mtr must also own x-latches to brothers of page, if those
brothers exist. NOTE: it is assumed that the caller has reserved enough
free extents so that the compression will always succeed if done!
@return whether compression occurred */
bool
btr_cur_compress_if_useful(
/*=======================*/
 btr_cur_t* cursor, /*!< in/out: cursor on the page to compress;
    cursor does not stay valid if !adjust and
compression occurs */
 bool  adjust, /*!< in: whether the cursor position should be
adjusted even when compression occurs */
 mtr_t*  mtr) /*!< in/out: mini-transaction */
{
 ut_ad(mtr->memo_contains_flagged(&cursor->index()->lock,
      MTR_MEMO_X_LOCK | MTR_MEMO_SX_LOCK));
 ut_ad(mtr->memo_contains_flagged(btr_cur_get_block(cursor),
      MTR_MEMO_PAGE_X_FIX));

 if (cursor->index()->is_spatial()) {
  const trx_t* trx = cursor->rtr_info->thr
   ? thr_get_trx(cursor->rtr_info->thr)
   : NULL;
  const buf_block_t* block = btr_cur_get_block(cursor);

  /* Check whether page lock prevents the compression */
  if (!lock_test_prdt_page_lock(trx, block->page.id())) {
   return(false);
  }
 }

 return btr_cur_compress_recommendation(cursor, mtr)
  && btr_compress(cursor, adjust, mtr) == DB_SUCCESS;
}

/*******************************************************//**
Removes the record on which the tree cursor is positioned on a leaf page.
It is assumed that the mtr has an x-latch on the page where the cursor is
positioned, but no latch on the whole tree.
@return error code
@retval DB_FAIL if the page would become too empty */
dberr_t
btr_cur_optimistic_delete(
/*======================*/
 btr_cur_t* cursor, /*!< in: cursor on leaf page, on the record to
    delete; cursor stays valid: if deletion
    succeeds, on function exit it points to the
successor of the deleted record */
 ulint  flags, /*!< in: BTR_CREATE_FLAG or 0 */
 mtr_t*  mtr) /*!< in: mtr; if this function returns
    TRUE on a leaf page of a secondary
    index, the mtr must be committed
before latching any further pages */
{
 buf_block_t* block;
 rec_t*  rec;
 mem_heap_t* heap  = NULL;
 rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
 rec_offs* offsets  = offsets_;
 rec_offs_init(offsets_);

 ut_ad(flags == 0 || flags == BTR_CREATE_FLAG);
 ut_ad(mtr->memo_contains_flagged(btr_cur_get_block(cursor),
      MTR_MEMO_PAGE_X_FIX));
 ut_ad(mtr->is_named_space(cursor->index()->table->space));
 ut_ad(!cursor->index()->is_dummy);

 /* This is intended only for leaf page deletions */

 block = btr_cur_get_block(cursor);

 ut_ad(block->page.id().space() == cursor->index()->table->space->id);
 ut_ad(page_is_leaf(buf_block_get_frame(block)));
 ut_ad(!dict_index_is_online_ddl(cursor->index())
       || cursor->index()->is_clust()
       || (flags & BTR_CREATE_FLAG));

 rec = btr_cur_get_rec(cursor);

 offsets = rec_get_offsets(rec, cursor->index(), offsets,
      cursor->index()->n_core_fields,
      ULINT_UNDEFINED, &heap);

 dberr_t err = DB_SUCCESS;
 DBUG_EXECUTE_IF("btr_force_pessimistic_delete",
  err = DB_FAIL; goto func_exit;);

 if (rec_offs_any_extern(offsets)
     || !btr_cur_can_delete_without_compress(cursor,
          rec_offs_size(offsets),
          mtr)) {
  /* prefetch siblings of the leaf for the pessimistic
operation. */
  btr_cur_prefetch_siblings(block, cursor->index(), mtr->trx);
  err = DB_FAIL;
  goto func_exit;
 }

 if (UNIV_UNLIKELY(block->page.id().page_no() == cursor->index()->page
     && page_get_n_recs(block->page.frame) == 1
     + (cursor->index()->is_instant()
        && !rec_is_metadata(rec, *cursor->index()))
     && !cursor->index()
     ->must_avoid_clear_instant_add())) {
  /* The whole index (and table) becomes logically empty.
  Empty the whole page. That is, if we are deleting the
  only user record, also delete the metadata record
  if one exists for instant ADD COLUMN (not generic ALTER TABLE).
  If we are deleting the metadata record and the
table becomes empty, clean up the whole page. */
  dict_index_t* index = cursor->index();
  const rec_t* first_rec = page_rec_get_next_const(
   page_get_infimum_rec(block->page.frame));
  if (UNIV_UNLIKELY(!first_rec)) {
   err = DB_CORRUPTION;
   goto func_exit;
  }
  ut_ad(!index->is_instant()
        || rec_is_metadata(first_rec, *index));
  const bool is_metadata = rec_is_metadata(rec, *index);
  /* We can remove the metadata when rolling back an
  instant ALTER TABLE operation, or when deleting the
  last user record on the page such that only metadata for
instant ADD COLUMN (not generic ALTER TABLE) remains. */
  const bool empty_table = is_metadata
   || !index->is_instant()
   || (first_rec != rec
       && rec_is_add_metadata(first_rec, *index));
  if (UNIV_LIKELY(empty_table)) {
   if (UNIV_LIKELY(!is_metadata && !flags)) {
    lock_update_delete(block, rec);
   }
   btr_page_empty(block, buf_block_get_page_zip(block),
           index, 0, mtr);
   if (index->is_instant()) {
    /* MDEV-17383: free metadata BLOBs! */
    index->clear_instant_alter();
   }

   page_cur_set_after_last(block,
      btr_cur_get_page_cur(cursor));
   goto func_exit;
  }
 }

 {
  if (UNIV_UNLIKELY(rec_get_info_bits(rec, page_is_comp(
           block->page.frame))
      & REC_INFO_MIN_REC_FLAG)) {
   /* This should be rolling back instant ADD COLUMN.
   If this is a recovered transaction, then
   index->is_instant() will hold until the
insert into SYS_COLUMNS is rolled back. */
   ut_ad(cursor->index()->table->supports_instant());
   ut_ad(cursor->index()->is_primary());
   ut_ad(!buf_block_get_page_zip(block));
   page_cur_delete_rec(btr_cur_get_page_cur(cursor),
         offsets, mtr);
   /* We must empty the PAGE_FREE list, because
   after rollback, this deleted metadata record
   would have too many fields, and we would be
unable to know the size of the freed record. */
   err = btr_page_reorganize(btr_cur_get_page_cur(cursor),
        mtr);
   goto func_exit;
  } else {
   if (!flags) {
    lock_update_delete(block, rec);
   }

   btr_search_update_hash_on_delete(cursor);
  }

  page_cur_delete_rec(btr_cur_get_page_cur(cursor),
        offsets, mtr);
 }

func_exit:
 if (UNIV_LIKELY_NULL(heap)) {
  mem_heap_free(heap);
 }

 return err;
}

/*************************************************************//**
Removes the record on which the tree cursor is positioned. Tries
to compress the page if its fillfactor drops below a threshold
or if it is the only page on the level. It is assumed that mtr holds
an x-latch on the tree and on the cursor page. To avoid deadlocks,
mtr must also own x-latches to brothers of page, if those brothers
exist.
@return TRUE if compression occurred and FALSE if not or something
wrong. */
ibool
btr_cur_pessimistic_delete(
/*=======================*/
 dberr_t* err, /*!< out: DB_SUCCESS or DB_OUT_OF_FILE_SPACE;
    the latter may occur because we may have
    to update node pointers on upper levels,
    and in the case of variable length keys
these may actually grow in size */
 ibool  has_reserved_extents, /*!< in: TRUE if the
    caller has already reserved enough free
    extents so that he knows that the operation
will succeed */
 btr_cur_t* cursor, /*!< in: cursor on the record to delete;
    if compression does not occur, the cursor
    stays valid: it points to successor of
deleted record on function exit */
 ulint  flags, /*!< in: BTR_CREATE_FLAG or 0 */
 bool  rollback,/*!< in: performing rollback? */
 mtr_t*  mtr) /*!< in: mtr */
{
 buf_block_t* block;
 page_t*  page;
 page_zip_des_t* page_zip;
 dict_index_t* index;
 rec_t*  rec;
 uint32_t n_reserved = 0;
 ibool  ret  = FALSE;
 mem_heap_t* heap;
 rec_offs* offsets;
#ifdef UNIV_DEBUG
 bool  parent_latched = false;
#endif /* UNIV_DEBUG */

 block = btr_cur_get_block(cursor);
 page = buf_block_get_frame(block);
 index = btr_cur_get_index(cursor);

 ut_d(++btr_cur_pessimistic_delete_calls);

 ut_ad(flags == 0 || flags == BTR_CREATE_FLAG);
 ut_ad(!dict_index_is_online_ddl(index)
       || dict_index_is_clust(index)
       || (flags & BTR_CREATE_FLAG));
 ut_ad(mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK
      | MTR_MEMO_SX_LOCK));
 ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
 ut_ad(mtr->is_named_space(index->table->space));
 ut_ad(!index->is_dummy);
 ut_ad(block->page.id().space() == index->table->space->id);

 if (!has_reserved_extents) {
  /* First reserve enough free space for the file segments
  of the index tree, so that the node pointer updates will
not fail because of lack of space */

  uint32_t n_extents = uint32_t(cursor->tree_height / 32 + 1);

  *err = fsp_reserve_free_extents(&n_reserved,
      index->table->space,
      n_extents,
      FSP_CLEANING, mtr);
  if (UNIV_UNLIKELY(*err != DB_SUCCESS)) {
   return(FALSE);
  }
 }

 heap = mem_heap_create(1024);
 rec = btr_cur_get_rec(cursor);
 page_zip = buf_block_get_page_zip(block);
#ifdef UNIV_ZIP_DEBUG
 ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */

 offsets = rec_get_offsets(rec, index, NULL, page_is_leaf(page)
      ? index->n_core_fields : 0,
      ULINT_UNDEFINED, &heap);

 if (rec_offs_any_extern(offsets)) {
  btr_rec_free_externally_stored_fields(index,
            rec, offsets, block,
            rollback, mtr);
#ifdef UNIV_ZIP_DEBUG
  ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */
 }

 rec_t* next_rec = NULL;
 bool min_mark_next_rec = false;

 if (page_is_leaf(page)) {
  const bool is_metadata = rec_is_metadata(
   rec, page_is_comp(block->page.frame));
  if (UNIV_UNLIKELY(is_metadata)) {
   /* This should be rolling back instant ALTER TABLE.
   If this is a recovered transaction, then
   index->is_instant() will hold until the
insert into SYS_COLUMNS is rolled back. */
   ut_ad(rollback);
   ut_ad(index->table->supports_instant());
   ut_ad(index->is_primary());
  } else if (flags == 0) {
   lock_update_delete(block, rec);
  }

  if (block->page.id().page_no() != index->page) {
   if (page_get_n_recs(page) < 2) {
    goto discard_page;
   }
  } else if (page_get_n_recs(page) == 1
      + (index->is_instant() && !is_metadata)
      && !index->must_avoid_clear_instant_add()) {
   /* The whole index (and table) becomes logically empty.
   Empty the whole page. That is, if we are deleting the
   only user record, also delete the metadata record
   if one exists for instant ADD COLUMN
   (not generic ALTER TABLE).
   If we are deleting the metadata record
   (in the rollback of instant ALTER TABLE) and the
table becomes empty, clean up the whole page. */

   const rec_t* first_rec = page_rec_get_next_const(
    page_get_infimum_rec(page));
   if (UNIV_UNLIKELY(!first_rec)) {
    *err = DB_CORRUPTION;
    goto err_exit;
   }
   ut_ad(!index->is_instant()
         || rec_is_metadata(first_rec, *index));
   if (is_metadata || !index->is_instant()
       || (first_rec != rec
    && rec_is_add_metadata(first_rec, *index))) {
    btr_page_empty(block, page_zip, index, 0, mtr);
    if (index->is_instant()) {
     /* MDEV-17383: free metadata BLOBs! */
     index->clear_instant_alter();
    }

    page_cur_set_after_last(
     block,
     btr_cur_get_page_cur(cursor));
    ret = TRUE;
    goto return_after_reservations;
   }
  }

  if (UNIV_LIKELY(!is_metadata)) {
   btr_search_update_hash_on_delete(cursor);
  } else {
   page_cur_delete_rec(btr_cur_get_page_cur(cursor),
         offsets, mtr);
   /* We must empty the PAGE_FREE list, because
   after rollback, this deleted metadata record
   would carry too many fields, and we would be
unable to know the size of the freed record. */
   *err = btr_page_reorganize(btr_cur_get_page_cur(cursor),
         mtr);
   ut_ad(!ret);
   goto err_exit;
  }
 } else if (UNIV_UNLIKELY(page_rec_is_first(rec, page))) {
  if (page_rec_is_last(rec, page)) {
discard_page:
   ut_ad(page_get_n_recs(page) == 1);
   /* If there is only one record, drop
the whole page. */

   btr_discard_page(cursor, mtr);

   ret = TRUE;
   goto return_after_reservations;
  }

  if (UNIV_UNLIKELY(!(next_rec = page_rec_get_next(rec)))) {
   ut_ad(!ret);
   *err = DB_CORRUPTION;
   goto err_exit;
  }

  btr_cur_t cur;
  cur.page_cur.index = index;
  cur.page_cur.block = block;

  if (!page_has_prev(page)) {
   /* If we delete the leftmost node pointer on a
   non-leaf level, we must mark the new leftmost node
pointer as the predefined minimum record */

   min_mark_next_rec = true;
  } else if (index->is_spatial()) {
   /* For rtree, if delete the leftmost node pointer,
we need to update parent page. */
   rtr_mbr_t father_mbr;
   rec_t*  father_rec;
   rec_offs* offsets;
   ulint  len;

   rtr_page_get_father_block(nullptr, heap, nullptr,
        &cur,
        cursor->rtr_info->thr, mtr);
   father_rec = btr_cur_get_rec(&cur);
   offsets = rec_get_offsets(father_rec, index, NULL,
        0, ULINT_UNDEFINED, &heap);

   rtr_read_mbr(rec_get_nth_field(
    father_rec, offsets, 0, &len), &father_mbr);

   rtr_update_mbr_field(&cur, offsets, NULL,
          page, &father_mbr, next_rec, mtr);
   ut_d(parent_latched = true);
  } else {
   /* Otherwise, if we delete the leftmost node pointer
   on a page, we have to change the parent node pointer
   so that it is equal to the new leftmost node pointer
on the page */
   ret = btr_page_get_father(mtr, &cur);
   if (!ret) {
    *err = DB_CORRUPTION;
    goto err_exit;
   }
   *err = btr_cur_node_ptr_delete(&cur, mtr);
   if (*err != DB_SUCCESS) {
got_err:
    ret = FALSE;
    goto err_exit;
   }

   const ulint level = btr_page_get_level(page);
   // FIXME: reuse the node_ptr from above
   dtuple_t* node_ptr = dict_index_build_node_ptr(
    index, next_rec, block->page.id().page_no(),
    heap, level);

   *err = btr_insert_on_non_leaf_level(
    flags, index, level + 1, node_ptr, mtr);
   if (*err != DB_SUCCESS) {
    ret = FALSE;
    goto got_err;
   }

   ut_d(parent_latched = true);
  }
 }

 /* SPATIAL INDEX never use U locks; we can allow page merges
 while holding X lock on the spatial index tree.
 Do not allow merges of non-leaf B-tree pages unless it is
safe to do so. */
 {
  const bool allow_merge = page_is_leaf(page)
   || dict_index_is_spatial(index)
   || btr_cur_will_modify_tree(
    index, page, BTR_INTENTION_DELETE, rec,
    btr_node_ptr_max_size(index),
    block->zip_size(), mtr);
  page_cur_delete_rec(btr_cur_get_page_cur(cursor),
        offsets, mtr);

  if (min_mark_next_rec) {
   btr_set_min_rec_mark(next_rec, *block, mtr);
  }

#ifdef UNIV_ZIP_DEBUG
  ut_a(!page_zip || page_zip_validate(page_zip, page, index));
#endif /* UNIV_ZIP_DEBUG */

  ut_ad(!parent_latched
        || btr_check_node_ptr(index, block,
         cursor->rtr_info
         ? cursor->rtr_info->thr
         : nullptr, mtr));

  if (!ret && btr_cur_compress_recommendation(cursor, mtr)) {
   if (UNIV_LIKELY(allow_merge)) {
    ret = btr_cur_compress_if_useful(
     cursor, FALSE, mtr);
   } else {
    ib::warn() << "Not merging page "
        << block->page.id()
        << " in index " << index->name
        << " of " << index->table->name;
    ut_ad("MDEV-14637" == 0);
   }
  }
 }

return_after_reservations:
 *err = DB_SUCCESS;
err_exit:
 mem_heap_free(heap);

#if 0 // FIXME: this used to be a no-op, and will cause trouble if enabled
 if (page_is_leaf(page)
     && !dict_index_is_online_ddl(index)) {
  mtr->release(index->lock);
  /* NOTE: We cannot release root block latch here, because it
has segment header and already modified in most of cases.*/
 }
#endif

 index->table->space->release_free_extents(n_reserved);
 return(ret);
}

/** Delete the node pointer in a parent page.
@param[in,out] parent cursor pointing to parent record
@param[in,out] mtr mini-transaction */
dberr_t btr_cur_node_ptr_delete(btr_cur_t* parent, mtr_t* mtr)
{
 ut_ad(mtr->memo_contains_flagged(btr_cur_get_block(parent),
      MTR_MEMO_PAGE_X_FIX));
 dberr_t err;
 ibool compressed = btr_cur_pessimistic_delete(&err, TRUE, parent,
            BTR_CREATE_FLAG, false,
            mtr);
 if (err == DB_SUCCESS && !compressed) {
  btr_cur_compress_if_useful(parent, FALSE, mtr);
 }

 return err;
}

/** Represents the cursor for the number of rows estimation. The
content is used for level-by-level diving and estimation the number of rows
on each level. */
class btr_est_cur_t
{
  /* Assume a page like:
  records:             (inf, a, b, c, d, sup)
  index of the record:    0, 1, 2, 3, 4, 5
  */

  /** Index of the record where the page cursor stopped on this level
  (index in alphabetical order). In the above example, if the search stopped on
  record 'c', then nth_rec will be 3. */
  ulint m_nth_rec;

  /** Number of the records on the page, not counting inf and sup.
  In the above example n_recs will be 4. */
  ulint m_n_recs;

  /** Search tuple */
  const dtuple_t &m_tuple;
  /** Cursor search mode */
  page_cur_mode_t m_mode;
  /** Page cursor which is used for search */
  page_cur_t m_page_cur;
  /** Page id of the page to get on level down, can differ from
  m_block->page.id at the moment when the child's page id is already found, but
  the child's block has not fetched yet */
  page_id_t m_page_id;
  /** Current block */
  buf_block_t *m_block;
  /** Page search mode, can differ from m_mode for non-leaf pages, see c-tor
  comments for details */
  page_cur_mode_t m_page_mode;

  /** Matched fields and bytes which are used for on-page search, see
  btr_cur_t::(up|low)_(match|bytes) comments for details */
  uint16_t m_up_match= 0;
  uint16_t m_up_bytes= 0;
  uint16_t m_low_match= 0;
  uint16_t m_low_bytes= 0;

public:
  btr_est_cur_t(dict_index_t *index, const dtuple_t &tuple,
                page_cur_mode_t mode)
      : m_tuple(tuple), m_mode(mode),
        m_page_id(index->table->space_id, index->page), m_block(nullptr)
  {

    ut_ad(dict_index_check_search_tuple(index, &tuple));
    ut_ad(dtuple_check_typed(&tuple));

    m_page_cur.index = index;
    /* We use these modified search modes on non-leaf levels of the B-tree.
    These let us end up in the right B-tree leaf. In that leaf we use the
    original search mode. */
    switch (mode) {
    case PAGE_CUR_GE:
      m_page_mode= PAGE_CUR_L;
      break;
    case PAGE_CUR_G:
      m_page_mode= PAGE_CUR_LE;
      break;
    default:
      ut_ad(mode == PAGE_CUR_L || mode == PAGE_CUR_LE);
      m_page_mode= mode;
      break;
    }
  }

  /** Retrieve block with m_page_id, release the previously gotten block
  if necessary. If this is a left border block cursor and both left and right
  border blocks have the same parent, don't unlatch the parent, as it must be
  latched to get the right block, and will be unlatched after the right block
  is fetched.
  @param  level distance from the leaf page level; ULINT_UNDEFINED when
          fetching the root page
  @param  mtr mtr
  @param  right_parent right border block parent, nullptr if the function
          is called for the right block itself
  @return true on success or false otherwise. */
  bool fetch_child(ulint level, mtr_t &mtr, const buf_block_t *right_parent)
  {
    buf_block_t *parent_block= m_block;

    m_block= btr_block_get(*index(), m_page_id.page_no(), RW_S_LATCH,
                           &mtr, nullptr);
    if (!m_block)
      return false;

    if (parent_block && parent_block != right_parent)
    {
      ut_ad(mtr.get_savepoint() >= 2);
      mtr.rollback_to_savepoint(1, 2);
    }

    return level == ULINT_UNDEFINED ||
      btr_page_get_level(m_block->page.frame) == level;
  }

  /** Sets page mode for leaves */
  void set_page_mode_for_leaves() { m_page_mode= m_mode; }

  /** Does search on the current page. If there is no border in m_tuple, then
  just move the cursor to the most left or right record.
  @param level current level on tree.
  @param root_height root height
  @param left true if this is left border, false otherwise.
  @return true on success, false otherwise. */
  bool search_on_page(ulint level, ulint root_height, bool left)
  {
    if (level != btr_page_get_level(m_block->page.frame))
      return false;

    m_n_recs= page_get_n_recs(m_block->page.frame);

    if (dtuple_get_n_fields(&m_tuple) > 0)
    {
      m_up_bytes= m_low_bytes= 0;
      m_page_cur.block= m_block;
      if (page_cur_search_with_match(&m_tuple, m_page_mode,
                                     &m_up_match, &m_low_match, &m_page_cur,
                                     nullptr))
        return false;
      m_nth_rec= page_rec_get_n_recs_before(page_cur_get_rec(&m_page_cur));
    }
    else if (left)
    {
      page_cur_set_before_first(m_block, &m_page_cur);
      if (level)
      {
        if (!page_cur_move_to_next(&m_page_cur))
          return false;
        m_nth_rec= 1;
      }
      else
        m_nth_rec= 0;
    }
    else
    {
      m_nth_rec= m_n_recs;
      if (!level)
      {
        page_cur_set_after_last(m_block, &m_page_cur);
        ++m_nth_rec;
      }
      else
      {
        m_page_cur.block= m_block;
        m_page_cur.rec= page_rec_get_nth(m_block->page.frame, m_nth_rec);
      }
    }

    return true;
  }

  /** Read page id of the current record child.
  @param offsets offsets array.
  @param heap heap for offsets array */
  void read_child_page_id(rec_offs **offsets, mem_heap_t **heap)
  {
    const rec_t *node_ptr= page_cur_get_rec(&m_page_cur);

    /* FIXME: get the child page number directly without computing offsets */
    *offsets= rec_get_offsets(node_ptr, index(), *offsets, 0, ULINT_UNDEFINED,
                              heap);

    /* Go to the child node */
    m_page_id.set_page_no(btr_node_ptr_get_child_page_no(node_ptr, *offsets));
  }

  /** @return true if left border should be counted */
  bool should_count_the_left_border() const
  {
    if (dtuple_get_n_fields(&m_tuple) > 0)
    {
      ut_ad(!page_rec_is_infimum(page_cur_get_rec(&m_page_cur)));
      return !page_rec_is_supremum(page_cur_get_rec(&m_page_cur));
    }
    ut_ad(page_rec_is_infimum(page_cur_get_rec(&m_page_cur)));
    return false;
  }

  /** @return true if right border should be counted */
  bool should_count_the_right_border() const
  {
    if (dtuple_get_n_fields(&m_tuple) > 0)
    {
      const rec_t *rec= page_cur_get_rec(&m_page_cur);
      ut_ad(!(m_mode == PAGE_CUR_L && page_rec_is_supremum(rec)));

      return (m_mode == PAGE_CUR_LE /* if the range is '<=' */
              /* and the record was found */
              && m_low_match >= dtuple_get_n_fields(&m_tuple)) ||
             (m_mode == PAGE_CUR_L /* or if the range is '<' */
              /* and there are any records to match the criteria, i.e. if the
              minimum record on the tree is 5 and x < 7 is specified then the
              cursor will be positioned at 5 and we should count the border,
              but if x < 2 is specified, then the cursor will be positioned at
              'inf' and we should not count the border */
              && !page_rec_is_infimum(rec));
      /* Notice that for "WHERE col <= 'foo'" the server passes to
      ha_innobase::records_in_range(): min_key=NULL (left-unbounded) which is
      expected max_key='foo' flag=HA_READ_AFTER_KEY (PAGE_CUR_G), which is
      unexpected - one would expect flag=HA_READ_KEY_OR_PREV (PAGE_CUR_LE). In
      this case the cursor will be positioned on the first record to the right
      of the requested one (can also be positioned on the 'sup') and we should
      not count the right border. */
    }
    ut_ad(page_rec_is_supremum(page_cur_get_rec(&m_page_cur)));

    /* The range specified is without a right border, just 'x > 123'
    or 'x >= 123' and search_on_page() positioned the cursor on the
    supremum record on the rightmost page, which must not be counted. */
    return false;
  }

  /** @return index */
  const dict_index_t *index() const { return m_page_cur.index; }

  /** @return current block */
  const buf_block_t *block() const { return m_block; }

  /** @return current page id */
  page_id_t page_id() const { return m_page_id; }

  /** Copies block pointer and savepoint from another btr_est_cur_t in the case
  if both left and right border cursors point to the same block.
  @param o reference to the other btr_est_cur_t object. */
  void set_block(const btr_est_cur_t &o) { m_block= o.m_block; }

  /** @return current record number. */
  ulint nth_rec() const { return m_nth_rec; }

  /** @return number of records in the current page. */
  ulint n_recs() const { return m_n_recs; }
};

/** Estimate the number of rows between the left record of the path and the
right one(non-inclusive) for the certain level on a B-tree. This function
starts from the page next to the left page and reads a few pages to the right,
counting their records. If we reach the right page quickly then we know exactly
how many records there are between left and right records and we set
is_n_rows_exact to true. After some page is latched, the previous page is
unlatched. If we cannot reach the right page quickly then we calculate the
average number of records in the pages scanned so far and assume that all pages
that we did not scan up to the right page contain the same number of records,
then we multiply that average to the number of pages between right and left
records (which is n_rows_on_prev_level). In this case we set is_n_rows_exact to
false.
@param level current level.
@param left_cur the cursor of the left page.
@param right_page_no right page number.
@param n_rows_on_prev_level number of rows on the previous level.
@param[out] is_n_rows_exact true if exact rows number is returned.
@param[in,out] mtr mtr,
@return number of rows, not including the borders (exact or estimated). */
static ha_rows btr_estimate_n_rows_in_range_on_level(
    ulint level, btr_est_cur_t &left_cur, uint32_t right_page_no,
    ha_rows n_rows_on_prev_level, bool &is_n_rows_exact, mtr_t &mtr)
{
  ha_rows n_rows= 0;
  uint n_pages_read= 0;
  /* Do not read more than this number of pages in order not to hurt
  performance with this code which is just an estimation. If we read this many
  pages before reaching right_page_no, then we estimate the average from the
  pages scanned so far. */
  static constexpr uint n_pages_read_limit= 9;
  buf_block_t *block= nullptr;
  const dict_index_t *index= left_cur.index();

  /* Assume by default that we will scan all pages between left and right(non
  inclusive) pages */
  is_n_rows_exact= true;

  /* Add records from the left page which are to the right of the record which
  serves as a left border of the range, if any (we don't include the record
  itself in this count). */
  if (left_cur.nth_rec() <= left_cur.n_recs())
  {
    n_rows+= left_cur.n_recs() - left_cur.nth_rec();
  }

  /* Count the records in the pages between left and right (non inclusive)
  pages */

  const fil_space_t *space= index->table->space;
  page_id_t page_id(space->id,
                    btr_page_get_next(buf_block_get_frame(left_cur.block())));

  if (page_id.page_no() == FIL_NULL)
    goto inexact;

  do
  {
    page_t *page;
    buf_block_t *prev_block= block;

    /* Fetch the page. */
    block= btr_block_get(*index, page_id.page_no(), RW_S_LATCH, &mtr, nullptr);

    if (prev_block)
    {
      ulint savepoint = mtr.get_savepoint();
      /* Index s-lock, p1, p2 latches, can also be p1 and p2 parent latch if
      they are not diverged */
      ut_ad(savepoint >= 3);
      mtr.rollback_to_savepoint(savepoint - 2, savepoint - 1);
    }

    if (!block || btr_page_get_level(buf_block_get_frame(block)) != level)
      goto inexact;

    page= buf_block_get_frame(block);

    /* It is possible but highly unlikely that the page was originally written
    by an old version of InnoDB that did not initialize FIL_PAGE_TYPE on other
    than B-tree pages. For example, this could be an almost-empty BLOB page
    that happens to contain the magic values in the fields
    that we checked above. */

    n_pages_read++;

    n_rows+= page_get_n_recs(page);

    page_id.set_page_no(btr_page_get_next(page));

    if (n_pages_read == n_pages_read_limit)
    {
      /* We read too many pages or we reached the end of the level
      without passing through right_page_no. */
      goto inexact;
    }

  } while (page_id.page_no() != right_page_no);

  if (block)
  {
    ut_ad(block == mtr.at_savepoint(mtr.get_savepoint() - 1));
    mtr.rollback_to_savepoint(mtr.get_savepoint() - 1);
  }

  return (n_rows);

inexact:

  if (block)
  {
    ut_ad(block == mtr.at_savepoint(mtr.get_savepoint() - 1));
    mtr.rollback_to_savepoint(mtr.get_savepoint() - 1);
  }

  is_n_rows_exact= false;

  /* We did interrupt before reaching right page */

  if (n_pages_read > 0)
  {
    /* The number of pages on this level is
    n_rows_on_prev_level, multiply it by the
    average number of recs per page so far */
    n_rows= n_rows_on_prev_level * n_rows / n_pages_read;
  }
  else
  {
    n_rows= 10;
  }

  return (n_rows);
}

/** Estimates the number of rows in a given index range. Do search in the
left page, then if there are pages between left and right ones, read a few
pages to the right, if the right page is reached, fetch it and count the exact
number of rows, otherwise count the estimated(see
btr_estimate_n_rows_in_range_on_level() for details) number if rows, and
fetch the right page. If leaves are reached, unlatch non-leaf pages except
the right leaf parent. After the right leaf page is fetched, commit mtr.
@param trx transaction
@param index B-tree
@param range_start first key
@param range_end   last key
@return estimated number of rows; */
ha_rows btr_estimate_n_rows_in_range(trx_t *trx, dict_index_t *index,
                                     btr_pos_t *range_start,
                                     btr_pos_t *range_end)
{
  DBUG_ENTER("btr_estimate_n_rows_in_range");

  if (UNIV_UNLIKELY(index->page == FIL_NULL || index->is_corrupted()))
    DBUG_RETURN(0);

  ut_ad(index->is_btree());

  mtr_t mtr{trx};
  btr_est_cur_t p1(index, *range_start->tuple, range_start->mode);
  btr_est_cur_t p2(index, *range_end->tuple, range_end->mode);

  ulint height;
  ulint root_height= 0; /* remove warning */

  mem_heap_t *heap= NULL;
  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets= offsets_;
  rec_offs_init(offsets_);

  mtr.start();

  ut_ad(mtr.get_savepoint() == 0);
  mtr_s_lock_index(index, &mtr);

  ha_rows table_n_rows= dict_table_get_n_rows(index->table);

  height= ULINT_UNDEFINED;

  /* This becomes true when the two paths do not pass through the same pages
  anymore. */
  bool diverged= false;
  /* This is the height, i.e. the number of levels from the root, where paths
   are not the same or adjacent any more. */
  ulint divergence_height= ULINT_UNDEFINED;
  bool should_count_the_left_border= true;
  bool should_count_the_right_border= true;
  bool is_n_rows_exact= true;
  ha_rows n_rows= 0;

  /* Loop and search until we arrive at the desired level. */
search_loop:
  if (!p1.fetch_child(height, mtr, p2.block()))
    goto error;

  if (height == ULINT_UNDEFINED)
  {
    /* We are in the root node */
    height= btr_page_get_level(buf_block_get_frame(p1.block()));
    root_height= height;
  }

  if (!height)
  {
    p1.set_page_mode_for_leaves();
    p2.set_page_mode_for_leaves();
  }

  if (p1.page_id() == p2.page_id())
    p2.set_block(p1);
  else
  {
    ut_ad(diverged);
    if (divergence_height != ULINT_UNDEFINED) {
      /* We need to call p1.search_on_page() here as
      btr_estimate_n_rows_in_range_on_level() uses p1.m_n_recs and
      p1.m_nth_rec. */
      if (!p1.search_on_page(height, root_height, true))
        goto error;
      n_rows= btr_estimate_n_rows_in_range_on_level(
          height, p1, p2.page_id().page_no(), n_rows, is_n_rows_exact, mtr);
    }
    if (!p2.fetch_child(height, mtr, nullptr))
      goto error;
  }

  if (height == 0)
    /* There is no need to release non-leaf pages here as they must already be
    unlatched in btr_est_cur_t::fetch_child(). Try to search on pages after
    releasing the index latch, to decrease contention. */
    mtr.rollback_to_savepoint(0, 1);

  /* There is no need to search on left page if
  divergence_height != ULINT_UNDEFINED, as it was already searched before
  btr_estimate_n_rows_in_range_on_level() call */
  if (divergence_height == ULINT_UNDEFINED &&
      !p1.search_on_page(height, root_height, true))
    goto error;

  if (!p2.search_on_page(height, root_height, false))
    goto error;

  if (!diverged && (p1.nth_rec() != p2.nth_rec()))
  {
    ut_ad(p1.page_id() == p2.page_id());
    diverged= true;
    if (p1.nth_rec() < p2.nth_rec())
    {
      /* We do not count the borders (nor the left nor the right one), thus
      "- 1". */
      n_rows= p2.nth_rec() - p1.nth_rec() - 1;

      if (n_rows > 0)
      {
        /* There is at least one row between the two borders pointed to by p1
        and p2, so on the level below the slots will point to non-adjacent
        pages. */
        divergence_height= root_height - height;
      }
    }
    else
    {
      /* It is possible that p1->nth_rec > p2->nth_rec if, for example, we have
      a single page tree which contains (inf, 5, 6, supr) and we select where x
      > 20 and x < 30; in this case p1->nth_rec will point to the supr record
      and p2->nth_rec will point to 6. */
      n_rows= 0;
      should_count_the_left_border= false;
      should_count_the_right_border= false;
    }
  }
  else if (diverged && divergence_height == ULINT_UNDEFINED)
  {

    if (p1.nth_rec() < p1.n_recs() || p2.nth_rec() > 1)
    {
      ut_ad(p1.page_id() != p2.page_id());
      divergence_height= root_height - height;

      n_rows= 0;

      if (p1.nth_rec() < p1.n_recs())
      {
        n_rows+= p1.n_recs() - p1.nth_rec();
      }

      if (p2.nth_rec() > 1)
      {
        n_rows+= p2.nth_rec() - 1;
      }
    }
  }
  else if (divergence_height != ULINT_UNDEFINED)
  {
    /* All records before the right page was already counted. Add records from
    p2->page_no which are to the left of the record which servers as a right
    border of the range, if any (we don't include the record itself in this
    count). */
    if (p2.nth_rec() > 1)
      n_rows+= p2.nth_rec() - 1;
  }

  if (height)
  {
    ut_ad(height > 0);
    height--;
    ut_ad(mtr.memo_contains(p1.index()->lock, MTR_MEMO_S_LOCK));
    ut_ad(mtr.memo_contains_flagged(p1.block(), MTR_MEMO_PAGE_S_FIX));
    p1.read_child_page_id(&offsets, &heap);
    ut_ad(mtr.memo_contains(p2.index()->lock, MTR_MEMO_S_LOCK));
    ut_ad(mtr.memo_contains_flagged(p2.block(), MTR_MEMO_PAGE_S_FIX));
    p2.read_child_page_id(&offsets, &heap);
    goto search_loop;
  }

  should_count_the_left_border=
      should_count_the_left_border && p1.should_count_the_left_border();
  should_count_the_right_border=
      should_count_the_right_border && p2.should_count_the_right_border();

  mtr.commit();
  if (UNIV_LIKELY_NULL(heap))
    mem_heap_free(heap);


  range_start->page_id= p1.page_id();
  range_end->page_id= p2.page_id();

  /* Here none of the borders were counted. For example, if on the leaf level
  we descended to:
  (inf, a, b, c, d, e, f, sup)
           ^        ^
         path1    path2
  then n_rows will be 2 (c and d). */

  if (is_n_rows_exact)
  {
    /* Only fiddle to adjust this off-by-one if the number is exact, otherwise
    we do much grosser adjustments below. */

    /* If both paths end up on the same record on the leaf level. */
    if (p1.page_id() == p2.page_id() && p1.nth_rec() == p2.nth_rec())
    {

      /* n_rows can be > 0 here if the paths were first different and then
      converged to the same record on the leaf level.
      For example:
      SELECT ... LIKE 'wait/synch/rwlock%'
      mode1=PAGE_CUR_GE,
      tuple1="wait/synch/rwlock"
      path1[0]={nth_rec=58, n_recs=58,
                page_no=3, page_level=1}
      path1[1]={nth_rec=56, n_recs=55,
                page_no=119, page_level=0}

      mode2=PAGE_CUR_G
      tuple2="wait/synch/rwlock"
      path2[0]={nth_rec=57, n_recs=57,
                page_no=3, page_level=1}
      path2[1]={nth_rec=56, n_recs=55,
                page_no=119, page_level=0} */

      /* If the range is such that we should count both borders, then avoid
      counting that record twice - once as a left border and once as a right
      border. Some of the borders should not be counted, e.g. [3,3). */
      n_rows= should_count_the_left_border && should_count_the_right_border;
    }
    else
      n_rows+= should_count_the_left_border + should_count_the_right_border;
  }

  if (root_height > divergence_height && !is_n_rows_exact)
    /* In trees whose height is > 1 our algorithm tends to underestimate:
    multiply the estimate by 2: */
    n_rows*= 2;

  DBUG_EXECUTE_IF("bug14007649", DBUG_RETURN(n_rows););

#ifdef NOT_USED
  /* Do not estimate the number of rows in the range to over 1 / 2 of the
  estimated rows in the whole table */

  if (n_rows > table_n_rows / 2 && !is_n_rows_exact)
  {

    n_rows= table_n_rows / 2;

    /* If there are just 0 or 1 rows in the table, then we estimate all rows
    are in the range */

    if (n_rows == 0)
      n_rows= table_n_rows;
  }
#else
                        if (n_rows > table_n_rows)
                          n_rows= table_n_rows;
#endif

  DBUG_RETURN(n_rows);

error:
  mtr.commit();
  if (UNIV_LIKELY_NULL(heap))
    mem_heap_free(heap);

  DBUG_RETURN(0);
}

/*================== EXTERNAL STORAGE OF BIG FIELDS ===================*/

/***********************************************************//**
Gets the offset of the pointer to the externally stored part of a field.
@return offset of the pointer to the externally stored part */
static
ulint
btr_rec_get_field_ref_offs(
/*=======================*/
 const rec_offs* offsets,/*!< in: array returned by rec_get_offsets() */
 ulint  n) /*!< in: index of the external field */
{
 ulint field_ref_offs;
 ulint local_len;

 ut_a(rec_offs_nth_extern(offsets, n));
 field_ref_offs = rec_get_nth_field_offs(offsets, n, &local_len);
 ut_a(len_is_stored(local_len));
 ut_a(local_len >= BTR_EXTERN_FIELD_REF_SIZE);

 return(field_ref_offs + local_len - BTR_EXTERN_FIELD_REF_SIZE);
}

/** Gets a pointer to the externally stored part of a field.
@param rec record
@param offsets rec_get_offsets(rec)
@param n index of the externally stored field
@return pointer to the externally stored part */
#define btr_rec_get_field_ref(rec, offsets, n)   \
 ((rec) + btr_rec_get_field_ref_offs(offsets, n))

/** Gets the externally stored size of a record, in units of a database page.
@param[in] rec record
@param[in] offsets array returned by rec_get_offsets()
@return externally stored part, in units of a database page */
ulint
btr_rec_get_externally_stored_len(
 const rec_t* rec,
 const rec_offs* offsets)
{
 ulint n_fields;
 ulint total_extern_len = 0;
 ulint i;

 ut_ad(!rec_offs_comp(offsets) || !rec_get_node_ptr_flag(rec));

 if (!rec_offs_any_extern(offsets)) {
  return(0);
 }

 n_fields = rec_offs_n_fields(offsets);

 for (i = 0; i < n_fields; i++) {
  if (rec_offs_nth_extern(offsets, i)) {

   ulint extern_len = mach_read_from_4(
    btr_rec_get_field_ref(rec, offsets, i)
    + BTR_EXTERN_LEN + 4);

   total_extern_len += ut_calc_align(
    extern_len, ulint(srv_page_size));
  }
 }

 return total_extern_len >> srv_page_size_shift;
}

/*******************************************************************//**
Sets the ownership bit of an externally stored field in a record. */
static
void
btr_cur_set_ownership_of_extern_field(
/*==================================*/
 buf_block_t* block, /*!< in/out: index page */
 rec_t*  rec, /*!< in/out: clustered index record */
 dict_index_t* index, /*!< in: index of the page */
 const rec_offs* offsets,/*!< in: array returned by rec_get_offsets() */
 ulint  i, /*!< in: field number */
 bool  val, /*!< in: value to set */
 mtr_t*  mtr) /*!< in: mtr, or NULL if not logged */
{
 byte* data;
 ulint local_len;
 ulint byte_val;

 data = rec_get_nth_field(rec, offsets, i, &local_len);
 ut_ad(rec_offs_nth_extern(offsets, i));
 ut_a(local_len >= BTR_EXTERN_FIELD_REF_SIZE);

 local_len -= BTR_EXTERN_FIELD_REF_SIZE;

 byte_val = mach_read_from_1(data + local_len + BTR_EXTERN_LEN);

 if (val) {
  byte_val &= ~BTR_EXTERN_OWNER_FLAG;
 } else {
#if defined UNIV_DEBUG || defined UNIV_BLOB_LIGHT_DEBUG
  ut_a(!(byte_val & BTR_EXTERN_OWNER_FLAG));
#endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */
  byte_val |= BTR_EXTERN_OWNER_FLAG;
 }

 if (UNIV_LIKELY_NULL(block->page.zip.data)) {
  mach_write_to_1(data + local_len + BTR_EXTERN_LEN, byte_val);
  page_zip_write_blob_ptr(block, rec, index, offsets, i, mtr);
 } else {
  mtr->write<1,mtr_t::MAYBE_NOP>(*block, data + local_len
            + BTR_EXTERN_LEN, byte_val);
 }
}

/*******************************************************************//**
Marks non-updated off-page fields as disowned by this record. The ownership
must be transferred to the updated record which is inserted elsewhere in the
index tree. In purge only the owner of externally stored field is allowed
to free the field. */
void
btr_cur_disown_inherited_fields(
/*============================*/
 buf_block_t* block, /*!< in/out: index page */
 rec_t*  rec, /*!< in/out: record in a clustered index */
 dict_index_t* index, /*!< in: index of the page */
 const rec_offs* offsets,/*!< in: array returned by rec_get_offsets() */
 const upd_t* update, /*!< in: update vector */
 mtr_t*  mtr) /*!< in/out: mini-transaction */
{
 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(!rec_offs_comp(offsets) || !rec_get_node_ptr_flag(rec));
 ut_ad(rec_offs_any_extern(offsets));

 for (uint16_t i = 0; i < rec_offs_n_fields(offsets); i++) {
  if (rec_offs_nth_extern(offsets, i)
      && !upd_get_field_by_field_no(update, i, false)) {
   btr_cur_set_ownership_of_extern_field(
    block, rec, index, offsets, i, false, mtr);
  }
 }
}

/*******************************************************************//**
Marks all extern fields in a record as owned by the record. This function
should be called if the delete mark of a record is removed: a not delete
marked record always owns all its extern fields. */
static
void
btr_cur_unmark_extern_fields(
/*=========================*/
 buf_block_t* block, /*!< in/out: index page */
 rec_t*  rec, /*!< in/out: record in a clustered index */
 dict_index_t* index, /*!< in: index of the page */
 const rec_offs* offsets,/*!< in: array returned by rec_get_offsets() */
 mtr_t*  mtr) /*!< in: mtr, or NULL if not logged */
{
 ut_ad(!rec_offs_comp(offsets) || !rec_get_node_ptr_flag(rec));
 if (!rec_offs_any_extern(offsets)) {
  return;
 }

 const ulint n = rec_offs_n_fields(offsets);

 for (ulint i = 0; i < n; i++) {
  if (rec_offs_nth_extern(offsets, i)) {
   btr_cur_set_ownership_of_extern_field(
    block, rec, index, offsets, i, true, mtr);
  }
 }
}

/*******************************************************************//**
Returns the length of a BLOB part stored on the header page.
@return part length */
static
uint32_t
btr_blob_get_part_len(
/*==================*/
 const byte* blob_header) /*!< in: blob header */
{
 return(mach_read_from_4(blob_header + BTR_BLOB_HDR_PART_LEN));
}

/*******************************************************************//**
Returns the page number where the next BLOB part is stored.
@return page number or FIL_NULL if no more pages */
static
uint32_t
btr_blob_get_next_page_no(
/*======================*/
 const byte* blob_header) /*!< in: blob header */
{
 return(mach_read_from_4(blob_header + BTR_BLOB_HDR_NEXT_PAGE_NO));
}

/** Deallocate a buffer block that was reserved for a BLOB part.
@param block   buffer block
@param all     flag whether to remove a ROW_FORMAT=COMPRESSED page
@param mtr     mini-transaction to commit */
static void btr_blob_free(buf_block_t *block, bool all, mtr_t *mtr)
{
  ut_ad(mtr->memo_contains_flagged(block, MTR_MEMO_PAGE_X_FIX));
  block->page.fix();
#ifdef UNIV_DEBUG
  const page_id_t page_id{block->page.id()};
  buf_pool_t::hash_chain &chain= buf_pool.page_hash.cell_get(page_id.fold());
#endif
  mtr->commit();

  mysql_mutex_lock(&buf_pool.mutex);
  block->page.unfix();
  ut_ad(block->page.id() == page_id);
  ut_ad(&block->page == buf_pool.page_hash.get(page_id, chain));

  if (!buf_LRU_free_page(&block->page, all) && all && block->page.zip.data)
    /* Attempt to deallocate the redundant copy of the uncompressed page
    if the whole ROW_FORMAT=COMPRESSED block cannot be deallocated. */
    buf_LRU_free_page(&block->page, false);

  mysql_mutex_unlock(&buf_pool.mutex);
}

/** Helper class used while writing blob pages, during insert or update. */
struct btr_blob_log_check_t {
 /** Persistent cursor on a clusterex index record with blobs. */
 btr_pcur_t* m_pcur;
 /** Mini transaction holding the latches for m_pcur */
 mtr_t*  m_mtr;
 /** rec_get_offsets(rec, index); offset of clust_rec */
 const rec_offs* m_offsets;
 /** The block containing clustered record */
 buf_block_t** m_block;
 /** The clustered record pointer */
 rec_t**  m_rec;
 /** The blob operation code */
 enum blob_op m_op;

 /** Constructor
 @param[in] pcur  persistent cursor on a clustered
     index record with blobs.
 @param[in] mtr  mini-transaction holding latches for
     pcur.
 @param[in] offsets  offsets of the clust_rec
 @param[in,out] block  record block containing pcur record
 @param[in,out] rec  the clustered record pointer
@param[in] op the blob operation code */
 btr_blob_log_check_t(
  btr_pcur_t* pcur,
  mtr_t*  mtr,
  const rec_offs* offsets,
  buf_block_t** block,
  rec_t**  rec,
  enum blob_op op)
  : m_pcur(pcur),
    m_mtr(mtr),
    m_offsets(offsets),
    m_block(block),
    m_rec(rec),
    m_op(op)
 {
  ut_ad(rec_offs_validate(*m_rec, m_pcur->index(), m_offsets));
  ut_ad((*m_block)->page.frame == page_align(*m_rec));
  ut_ad(*m_rec == btr_pcur_get_rec(m_pcur));
 }

 /** Check if there is enough space in log file. Commit and re-start the
mini transaction. */
 void check()
 {
  dict_index_t* index = m_pcur->index();
  ulint  offs = 0;
  uint32_t page_no = FIL_NULL;

  if (UNIV_UNLIKELY(m_op == BTR_STORE_INSERT_BULK)) {
   offs = *m_rec - (*m_block)->page.frame;
   ut_ad(offs == page_offset(*m_rec));
   page_no = (*m_block)->page.id().page_no();
   (*m_block)->page.fix();
   ut_ad(page_no != FIL_NULL);
  } else {
   btr_pcur_store_position(m_pcur, m_mtr);
  }
  m_mtr->commit();

  DEBUG_SYNC_C("blob_write_middle");

  const mtr_log_t log_mode = m_mtr->get_log_mode();
  m_mtr->start();
  m_mtr->set_log_mode(log_mode);
  index->set_modified(*m_mtr);

  log_free_check();

  DEBUG_SYNC_C("blob_write_middle_after_check");

  if (UNIV_UNLIKELY(page_no != FIL_NULL)) {
   dberr_t err;
   if (UNIV_LIKELY(index->page != page_no)) {
    ut_a(btr_root_block_get(index, RW_SX_LATCH,
       m_mtr, &err));
   }
   m_pcur->btr_cur.page_cur.block = btr_block_get(
    *index, page_no, RW_X_LATCH, m_mtr);
   /* The page should not be evicted or corrupted while
we are holding a buffer-fix on it. */
   m_pcur->btr_cur.page_cur.block->page.unfix();
   m_pcur->btr_cur.page_cur.rec
    = m_pcur->btr_cur.page_cur.block->page.frame
    + offs;
  } else {
   ut_ad(m_pcur->rel_pos == BTR_PCUR_ON);
   mtr_sx_lock_index(index, m_mtr);
   ut_a(m_pcur->restore_position(
         BTR_MODIFY_ROOT_AND_LEAF_ALREADY_LATCHED,
         m_mtr) == btr_pcur_t::SAME_ALL);
  }

  *m_block = btr_pcur_get_block(m_pcur);
  *m_rec  = btr_pcur_get_rec(m_pcur);

  rec_offs_make_valid(*m_rec, index, true,
        const_cast<rec_offs*>(m_offsets));

  ut_ad(m_mtr->memo_contains_page_flagged(
        *m_rec,
        MTR_MEMO_PAGE_X_FIX | MTR_MEMO_PAGE_SX_FIX));

  ut_ad((m_op == BTR_STORE_INSERT_BULK)
        == !m_mtr->memo_contains_flagged(&index->lock,
             MTR_MEMO_SX_LOCK
             | MTR_MEMO_X_LOCK));
 }
};

/*******************************************************************//**
Stores the fields in big_rec_vec to the tablespace and puts pointers to
them in rec.  The extern flags in rec will have to be set beforehand.
The fields are stored on pages allocated from leaf node
file segment of the index tree.

TODO: If the allocation extends the tablespace, it will not be redo logged, in
any mini-transaction.  Tablespace extension should be redo-logged, so that
recovery will not fail when the big_rec was written to the extended portion of
the file, in case the file was somehow truncated in the crash.

@return DB_SUCCESS or DB_OUT_OF_FILE_SPACE */
dberr_t
btr_store_big_rec_extern_fields(
/*============================*/
 btr_pcur_t* pcur,  /*!< in: a persistent cursor */
 rec_offs* offsets, /*!< in/out: rec_get_offsets() on
     pcur. the "external storage" flags
     in offsets will correctly correspond
to rec when this function returns */
 const big_rec_t*big_rec_vec, /*!< in: vector containing fields
to be stored externally */
 mtr_t*  btr_mtr, /*!< in/out: mtr containing the
     latches to the clustered index. can be
committed and restarted. */
 enum blob_op op)  /*! in: operation code */
{
 byte*  field_ref;
 ulint  extern_len;
 ulint  store_len;
 ulint  i;
 mtr_t  mtr{btr_mtr->trx};
 mem_heap_t* heap = NULL;
 page_zip_des_t* page_zip;
 z_stream c_stream;
 dberr_t  error  = DB_SUCCESS;
 dict_index_t* index  = pcur->index();
 buf_block_t* rec_block = btr_pcur_get_block(pcur);
 rec_t*  rec  = btr_pcur_get_rec(pcur);

 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(rec_offs_any_extern(offsets));
 ut_ad(op == BTR_STORE_INSERT_BULK
       || btr_mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK
      | MTR_MEMO_SX_LOCK));
 ut_ad(btr_mtr->memo_contains_flagged(rec_block, MTR_MEMO_PAGE_X_FIX));
 ut_ad(buf_block_get_frame(rec_block) == page_align(rec));
 ut_a(dict_index_is_clust(index));

 if (!fil_page_index_page_check(btr_pcur_get_page(pcur))) {
  if (op != BTR_STORE_INSERT_BULK) {
   return DB_PAGE_CORRUPTED;
  }
 }

 btr_blob_log_check_t redo_log(pcur, btr_mtr, offsets, &rec_block,
          &rec, op);
 page_zip = buf_block_get_page_zip(rec_block);

 if (page_zip) {
  int err;

  /* Zlib deflate needs 128 kilobytes for the default
  window size, plus 512 << memLevel, plus a few
  kilobytes for small objects.  We use reduced memLevel
  to limit the memory consumption, and preallocate the
heap, hoping to avoid memory fragmentation. */
  heap = mem_heap_create(250000);
  page_zip_set_alloc(&c_stream, heap);

  err = deflateInit2(&c_stream, int(page_zip_level),
       Z_DEFLATED, 15, 7, Z_DEFAULT_STRATEGY);
  ut_a(err == Z_OK);
 }

#if defined UNIV_DEBUG || defined UNIV_BLOB_LIGHT_DEBUG
 /* All pointers to externally stored columns in the record
 must either be zero or they must be pointers to inherited
columns, owned by this record or an earlier record version. */
 for (i = 0; i < big_rec_vec->n_fields; i++) {
  field_ref = btr_rec_get_field_ref(
   rec, offsets, big_rec_vec->fields[i].field_no);

  ut_a(!(field_ref[BTR_EXTERN_LEN] & BTR_EXTERN_OWNER_FLAG));
  /* Either this must be an update in place,
  or the BLOB must be inherited, or the BLOB pointer
must be zero (will be written in this function). */
  ut_a(op == BTR_STORE_UPDATE
       || (field_ref[BTR_EXTERN_LEN] & BTR_EXTERN_INHERITED_FLAG)
       || !memcmp(field_ref, field_ref_zero,
    BTR_EXTERN_FIELD_REF_SIZE));
 }
#endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */

 /* Space available in compressed page to carry blob data */
 const ulint payload_size_zip = rec_block->physical_size()
  - FIL_PAGE_DATA;

 /* Space available in uncompressed page to carry blob data */
 const ulint payload_size = payload_size_zip
  - (BTR_BLOB_HDR_SIZE + FIL_PAGE_DATA_END);

 /* We have to create a file segment to the tablespace
for each field and put the pointer to the field in rec */

 for (i = 0; i < big_rec_vec->n_fields; i++) {
  const ulint field_no = big_rec_vec->fields[i].field_no;

  field_ref = btr_rec_get_field_ref(rec, offsets, field_no);
#if defined UNIV_DEBUG || defined UNIV_BLOB_LIGHT_DEBUG
  /* A zero BLOB pointer should have been initially inserted. */
  ut_a(!memcmp(field_ref, field_ref_zero,
        BTR_EXTERN_FIELD_REF_SIZE));
#endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */
  extern_len = big_rec_vec->fields[i].len;
  MEM_CHECK_DEFINED(big_rec_vec->fields[i].data, extern_len);
  ut_a(extern_len > 0);

  uint32_t prev_page_no = FIL_NULL;

  if (page_zip) {
   int err = deflateReset(&c_stream);
   ut_a(err == Z_OK);

   c_stream.next_in = (Bytef*)
    big_rec_vec->fields[i].data;
   c_stream.avail_in = static_cast<uInt>(extern_len);
  }

  for (ulint blob_npages = 0;; ++blob_npages) {
   buf_block_t* block;
   const ulint commit_freq = 4;

   ut_ad(page_align(field_ref) == page_align(rec));

   if (!(blob_npages % commit_freq)) {

    redo_log.check();

    field_ref = btr_rec_get_field_ref(
     rec, offsets, field_no);

    page_zip = buf_block_get_page_zip(rec_block);
   }

   ut_ad(btr_mtr->get_already_latched(
          page_id_t{index->table->space_id, index->page},
          MTR_MEMO_PAGE_SX_FIX));

   mtr.start();
   index->set_modified(mtr);
   mtr.set_log_mode_sub(*btr_mtr);

   rec_block->page.fix();
   rec_block->page.lock.x_lock();

   mtr.memo_push(rec_block, MTR_MEMO_PAGE_X_FIX);
#ifdef BTR_CUR_HASH_ADAPT
   ut_ad(!btr_search_check_marked_free_index(rec_block));
#endif

   uint32_t hint_prev = prev_page_no;
   if (hint_prev == FIL_NULL) {
    hint_prev = rec_block->page.id().page_no();
   }

   block = btr_page_alloc(index, hint_prev + 1,
            FSP_NO_DIR, 0, &mtr, &mtr,
            &error);

   DBUG_EXECUTE_IF("btr_page_alloc_fail",
     block= nullptr;
     error= DB_OUT_OF_FILE_SPACE;);
   if (!block) {
alloc_fail:
                                mtr.commit();
    goto func_exit;
   }

   const uint32_t space_id = block->page.id().space();
   const uint32_t page_no = block->page.id().page_no();

   if (prev_page_no == FIL_NULL) {
   } else if (buf_block_t* prev_block =
       buf_page_get_gen(page_id_t(space_id,
         prev_page_no),
                                                    rec_block->zip_size(),
                                                    RW_X_LATCH, nullptr,
                                                    BUF_GET, &mtr, &error)) {
    if (page_zip) {
     mtr.write<4>(*prev_block,
           prev_block->page.frame
           + FIL_PAGE_NEXT,
           page_no);
     memcpy_aligned<4>(
      buf_block_get_page_zip(
       prev_block)
      ->data + FIL_PAGE_NEXT,
      prev_block->page.frame
      + FIL_PAGE_NEXT, 4);
    } else {
     mtr.write<4>(*prev_block,
           BTR_BLOB_HDR_NEXT_PAGE_NO
           + FIL_PAGE_DATA
           + prev_block->page.frame,
           page_no);
    }
   } else {
    goto alloc_fail;
   }

   ut_ad(!page_has_siblings(block->page.frame));
   ut_ad(!fil_page_get_type(block->page.frame));

   if (page_zip) {
    int  err;
    page_zip_des_t* blob_page_zip;

    mtr.write<1>(*block,
          FIL_PAGE_TYPE + 1
          + block->page.frame,
          prev_page_no == FIL_NULL
          ? FIL_PAGE_TYPE_ZBLOB
          : FIL_PAGE_TYPE_ZBLOB2);
    block->page.zip.data[FIL_PAGE_TYPE + 1]
     = block->page.frame[FIL_PAGE_TYPE + 1];

    c_stream.next_out = block->page.frame
     + FIL_PAGE_DATA;
    c_stream.avail_out = static_cast<uInt>(
     payload_size_zip);

    err = deflate(&c_stream, Z_FINISH);
    ut_a(err == Z_OK || err == Z_STREAM_END);
    ut_a(err == Z_STREAM_END
         || c_stream.avail_out == 0);

    mtr.memcpy(*block,
        FIL_PAGE_DATA,
        page_zip_get_size(page_zip)
        - FIL_PAGE_DATA
        - c_stream.avail_out);
    /* Copy the page to compressed storage,
    because it will be flushed to disk
from there. */
    blob_page_zip = buf_block_get_page_zip(block);
    ut_ad(blob_page_zip);
    ut_ad(page_zip_get_size(blob_page_zip)
          == page_zip_get_size(page_zip));
    memcpy(blob_page_zip->data, block->page.frame,
           page_zip_get_size(page_zip));

    if (err == Z_OK && prev_page_no != FIL_NULL) {

     goto next_zip_page;
    }

    if (err == Z_STREAM_END) {
     mach_write_to_4(field_ref
       + BTR_EXTERN_LEN, 0);
     mach_write_to_4(field_ref
       + BTR_EXTERN_LEN + 4,
       c_stream.total_in);
    } else {
     memset(field_ref + BTR_EXTERN_LEN,
            0, 8);
    }

    if (prev_page_no == FIL_NULL) {
     ut_ad(blob_npages == 0);
     mach_write_to_4(field_ref
       + BTR_EXTERN_SPACE_ID,
       space_id);

     mach_write_to_4(field_ref
       + BTR_EXTERN_PAGE_NO,
       page_no);

     mach_write_to_4(field_ref
       + BTR_EXTERN_OFFSET,
       FIL_PAGE_NEXT);
    }

    /* We compress a page when finish bulk insert.*/
    if (UNIV_LIKELY(op != BTR_STORE_INSERT_BULK)) {
     page_zip_write_blob_ptr(
      rec_block, rec, index, offsets,
      field_no, &mtr);
    }

next_zip_page:
    prev_page_no = page_no;

    /* Commit mtr and release the
uncompressed page frame to save memory. */
    btr_blob_free(block, FALSE, &mtr);

    if (err == Z_STREAM_END) {
     break;
    }
   } else {
    mtr.write<1>(*block, FIL_PAGE_TYPE + 1
          + block->page.frame,
          FIL_PAGE_TYPE_BLOB);

    if (extern_len > payload_size) {
     store_len = payload_size;
    } else {
     store_len = extern_len;
    }

    mtr.memcpy<mtr_t::MAYBE_NOP>(
     *block,
     FIL_PAGE_DATA + BTR_BLOB_HDR_SIZE
     + block->page.frame,
     static_cast<const byte*>
     (big_rec_vec->fields[i].data)
     + big_rec_vec->fields[i].len
     - extern_len, store_len);
    mtr.write<4>(*block, BTR_BLOB_HDR_PART_LEN
          + FIL_PAGE_DATA
          + block->page.frame,
          store_len);
    compile_time_assert(FIL_NULL == 0xffffffff);
    mtr.memset(block, BTR_BLOB_HDR_NEXT_PAGE_NO
        + FIL_PAGE_DATA, 4, 0xff);

    extern_len -= store_len;

    ut_ad(!mach_read_from_4(BTR_EXTERN_LEN
       + field_ref));
    mtr.write<4>(*rec_block,
          BTR_EXTERN_LEN + 4 + field_ref,
          big_rec_vec->fields[i].len
          - extern_len);

    if (prev_page_no == FIL_NULL) {
     ut_ad(blob_npages == 0);
     mtr.write<4,mtr_t::MAYBE_NOP>(
      *rec_block,
      field_ref + BTR_EXTERN_SPACE_ID,
      space_id);

     mtr.write<4>(*rec_block, field_ref
           + BTR_EXTERN_PAGE_NO,
           page_no);

     mtr.write<4>(*rec_block, field_ref
           + BTR_EXTERN_OFFSET,
           FIL_PAGE_DATA);
    }

    prev_page_no = page_no;

    mtr.commit();

    if (extern_len == 0) {
     break;
    }
   }
  }

  DBUG_EXECUTE_IF("btr_store_big_rec_extern",
    error = DB_OUT_OF_FILE_SPACE;
    goto func_exit;);

  rec_offs_make_nth_extern(offsets, field_no);
 }

func_exit:
 if (page_zip) {
  deflateEnd(&c_stream);
 }

 if (heap != NULL) {
  mem_heap_free(heap);
 }

#if defined UNIV_DEBUG || defined UNIV_BLOB_LIGHT_DEBUG
 /* All pointers to externally stored columns in the record
must be valid. */
 for (i = 0; i < rec_offs_n_fields(offsets); i++) {
  if (!rec_offs_nth_extern(offsets, i)) {
   continue;
  }

  field_ref = btr_rec_get_field_ref(rec, offsets, i);

  /* The pointer must not be zero if the operation
succeeded. */
  ut_a(0 != memcmp(field_ref, field_ref_zero,
     BTR_EXTERN_FIELD_REF_SIZE)
       || error != DB_SUCCESS);
  /* The column must not be disowned by this record. */
  ut_a(!(field_ref[BTR_EXTERN_LEN] & BTR_EXTERN_OWNER_FLAG));
 }
#endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */
 return(error);
}

/** Check the FIL_PAGE_TYPE on an uncompressed BLOB page.
@param block   uncompressed BLOB page
@param op      operation
@return whether the type is invalid */
static bool btr_check_blob_fil_page_type(const buf_block_t& block,
                                         const char *op)
{
  uint16_t type= fil_page_get_type(block.page.frame);

  if (UNIV_LIKELY(type == FIL_PAGE_TYPE_BLOB));
  else if (fil_space_t *space= fil_space_t::get(block.page.id().space()))
  {
    /* Old versions of InnoDB did not initialize FIL_PAGE_TYPE on BLOB
    pages.  Do not print anything about the type mismatch when reading
    a BLOB page that may be from old versions. */
    bool fail= space->full_crc32() || DICT_TF_HAS_ATOMIC_BLOBS(space->flags);
    if (fail)
      sql_print_error("InnoDB: FIL_PAGE_TYPE=%u on BLOB %s file %s page %u",
                      type, op, space->chain.start->name,
                      block.page.id().page_no());
    space->release();
    return fail;
  }
  return false;
}

/*******************************************************************//**
Frees the space in an externally stored field to the file space
management if the field in data is owned by the externally stored field,
in a rollback we may have the additional condition that the field must
not be inherited. */
void
btr_free_externally_stored_field(
/*=============================*/
 dict_index_t* index,  /*!< in: index of the data, the index
     tree MUST be X-latched; if the tree
     height is 1, then also the root page
     must be X-latched! (this is relevant
     in the case this function is called
     from purge where 'data' is located on
     an undo log page, not an index
page) */
 byte*  field_ref, /*!< in/out: field reference */
 const rec_t* rec,  /*!< in: record containing field_ref, for
page_zip_write_blob_ptr(), or NULL */
 const rec_offs* offsets, /*!< in: rec_get_offsets(rec, index),
or NULL */
 buf_block_t* block,  /*!< in/out: page of field_ref */
 ulint  i,  /*!< in: field number of field_ref;
ignored if rec == NULL */
 bool  rollback, /*!< in: performing rollback? */
 mtr_t*  local_mtr) /*!< in: mtr
     containing the latch to data an an
X-latch to the index tree */
{
 const uint32_t space_id = mach_read_from_4(
  field_ref + BTR_EXTERN_SPACE_ID);

 ut_ad(index->is_primary());
 ut_ad(block->page.lock.have_x());
 ut_ad(local_mtr->memo_contains_flagged(&index->lock, MTR_MEMO_X_LOCK
            | MTR_MEMO_SX_LOCK));
 ut_ad(local_mtr->memo_contains_page_flagged(field_ref,
          MTR_MEMO_PAGE_X_FIX));
 ut_ad(!rec || rec_offs_validate(rec, index, offsets));
 ut_ad(!rec || field_ref == btr_rec_get_field_ref(rec, offsets, i));
 ut_ad(index->table->space_id == index->table->space->id);
 ut_ad(local_mtr->is_named_space(index->table->space));

 if (UNIV_UNLIKELY(!memcmp(field_ref, field_ref_zero,
      BTR_EXTERN_FIELD_REF_SIZE))) {
  /* In the rollback, we may encounter a clustered index
  record with some unwritten off-page columns. There is
nothing to free then. */
  ut_a(rollback);
  return;
 }

 ut_ad(!(mach_read_from_4(field_ref + BTR_EXTERN_LEN)
         & ~((BTR_EXTERN_OWNER_FLAG
              | BTR_EXTERN_INHERITED_FLAG) << 24)));
 ut_ad(space_id == index->table->space_id);

 const ulint ext_zip_size = index->table->space->zip_size();
 /* !rec holds in a call from purge when field_ref is in an undo page */
 ut_ad(rec || !block->page.zip.data);

 for (mtr_t mtr{local_mtr->trx};;) {
  mtr.start();
  mtr.set_spaces(*local_mtr);
  mtr.set_log_mode_sub(*local_mtr);

  ut_ad(!index->table->is_temporary()
        || local_mtr->get_log_mode() == MTR_LOG_NO_REDO);

  const uint32_t page_no = mach_read_from_4(
   field_ref + BTR_EXTERN_PAGE_NO);
  buf_block_t* ext_block;

  if (/* There is no external storage data */
      page_no == FIL_NULL
      /* This field does not own the externally stored field */
      || (mach_read_from_1(field_ref + BTR_EXTERN_LEN)
   & BTR_EXTERN_OWNER_FLAG)
      /* Rollback and inherited field */
      || (rollback
   && (mach_read_from_1(field_ref + BTR_EXTERN_LEN)
       & BTR_EXTERN_INHERITED_FLAG))) {
skip_free:
   /* Do not free */
   mtr.commit();

   return;
  }

  ext_block = buf_page_get(page_id_t(space_id, page_no),
      ext_zip_size, RW_X_LATCH, &mtr);

  if (!ext_block) {
   goto skip_free;
  }

  /* The buffer pool block containing the BLOB pointer is
  exclusively latched by local_mtr. To satisfy some design
constraints, we must recursively latch it in mtr as well. */
  block->fix();
  block->page.lock.x_lock();

  mtr.memo_push(block, MTR_MEMO_PAGE_X_FIX);
#ifdef BTR_CUR_HASH_ADAPT
  ut_ad(!btr_search_check_marked_free_index(block));
#endif

  const page_t* page = buf_block_get_frame(ext_block);

  if (ext_zip_size) {
   /* Note that page_zip will be NULL
in row_purge_upd_exist_or_extern(). */
   switch (fil_page_get_type(page)) {
   case FIL_PAGE_TYPE_ZBLOB:
   case FIL_PAGE_TYPE_ZBLOB2:
    break;
   default:
    MY_ASSERT_UNREACHABLE();
   }
   const uint32_t next_page_no = mach_read_from_4(
    page + FIL_PAGE_NEXT);

   btr_page_free(index, ext_block, &mtr, true,
          local_mtr->memo_contains(
           *index->table->space));

   if (UNIV_LIKELY_NULL(block->page.zip.data)) {
    mach_write_to_4(field_ref + BTR_EXTERN_PAGE_NO,
      next_page_no);
    memset(field_ref + BTR_EXTERN_LEN + 4, 0, 4);
    page_zip_write_blob_ptr(block, rec, index,
       offsets, i, &mtr);
   } else {
    mtr.write<4>(*block,
          BTR_EXTERN_PAGE_NO + field_ref,
          next_page_no);
    mtr.write<4,mtr_t::MAYBE_NOP>(*block,
             BTR_EXTERN_LEN
             + 4 + field_ref,
             0U);
   }
  } else {
   ut_ad(!block->page.zip.data);
   btr_check_blob_fil_page_type(*ext_block, "purge");

   const uint32_t next_page_no = mach_read_from_4(
    page + FIL_PAGE_DATA
    + BTR_BLOB_HDR_NEXT_PAGE_NO);
   btr_page_free(index, ext_block, &mtr, true,
          local_mtr->memo_contains(
           *index->table->space));

   mtr.write<4>(*block, BTR_EXTERN_PAGE_NO + field_ref,
         next_page_no);
   /* Zero out the BLOB length.  If the server
   crashes during the execution of this function,
   trx_rollback_all_recovered() could
   dereference the half-deleted BLOB, fetching a
wrong prefix for the BLOB. */
   mtr.write<4,mtr_t::MAYBE_NOP>(*block,
            BTR_EXTERN_LEN + 4
            + field_ref, 0U);
  }

  /* Commit mtr and release the BLOB block to save memory. */
  btr_blob_free(ext_block, TRUE, &mtr);
 }
}

/***********************************************************//**
Frees the externally stored fields for a record. */
static
void
btr_rec_free_externally_stored_fields(
/*==================================*/
 dict_index_t* index, /*!< in: index of the data, the index
tree MUST be X-latched */
 rec_t*  rec, /*!< in/out: record */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec, index) */
 buf_block_t* block, /*!< in: index page of rec */
 bool  rollback,/*!< in: performing rollback? */
 mtr_t*  mtr) /*!< in: mini-transaction handle which contains
    an X-latch to record page and to the index
tree */
{
 ulint n_fields;
 ulint i;

 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(mtr->memo_contains_page_flagged(rec, MTR_MEMO_PAGE_X_FIX));
 ut_ad(index->is_primary());
 ut_ad(page_rec_is_leaf(rec));
 /* Free possible externally stored fields in the record */

 ut_ad(dict_table_is_comp(index->table) == !!rec_offs_comp(offsets));
 n_fields = rec_offs_n_fields(offsets);

 for (i = 0; i < n_fields; i++) {
  if (rec_offs_nth_extern(offsets, i)) {
   btr_free_externally_stored_field(
    index, btr_rec_get_field_ref(rec, offsets, i),
    rec, offsets, block, i, rollback, mtr);
  }
 }
}

/***********************************************************//**
Frees the externally stored fields for a record, if the field is mentioned
in the update vector. */
static
void
btr_rec_free_updated_extern_fields(
/*===============================*/
 dict_index_t* index, /*!< in: index of rec; the index tree MUST be
X-latched */
 rec_t*  rec, /*!< in/out: record */
 buf_block_t* block, /*!< in: index page of rec */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec, index) */
 const upd_t* update, /*!< in: update vector */
 bool  rollback,/*!< in: performing rollback? */
 mtr_t*  mtr) /*!< in: mini-transaction handle which contains
an X-latch to record page and to the tree */
{
 ulint n_fields;
 ulint i;

 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(mtr->memo_contains_page_flagged(rec, MTR_MEMO_PAGE_X_FIX));

 /* Free possible externally stored fields in the record */

 n_fields = upd_get_n_fields(update);

 for (i = 0; i < n_fields; i++) {
  const upd_field_t* ufield = upd_get_nth_field(update, i);

  if (rec_offs_nth_extern(offsets, ufield->field_no)) {
   ulint len;
   byte* data = rec_get_nth_field(
    rec, offsets, ufield->field_no, &len);
   ut_a(len >= BTR_EXTERN_FIELD_REF_SIZE);

   btr_free_externally_stored_field(
    index, data + len - BTR_EXTERN_FIELD_REF_SIZE,
    rec, offsets, block,
    ufield->field_no, rollback, mtr);
  }
 }
}

/*******************************************************************//**
Copies the prefix of an uncompressed BLOB.  The clustered index record
that points to this BLOB must be protected by a lock or a page latch.
@return number of bytes written to buf */
static
ulint
btr_copy_blob_prefix(
/*=================*/
 byte*  buf, /*!< out: the externally stored part of
the field, or a prefix of it */
 uint32_t len, /*!< in: length of buf, in bytes */
 page_id_t id, /*!< in: page identifier of the first BLOB page */
 uint32_t offset) /*!< in: offset on the first BLOB page */
{
 ulint copied_len = 0;
 THD* thd{current_thd};

 for (mtr_t mtr{thd ? thd_to_trx(thd) : nullptr};;) {
  buf_block_t* block;
  const page_t* page;
  const byte* blob_header;
  ulint  part_len;
  ulint  copy_len;

  mtr_start(&mtr);

  block = buf_page_get(id, 0, RW_S_LATCH, &mtr);
  if (!block || btr_check_blob_fil_page_type(*block, "read")) {
   mtr.commit();
   return copied_len;
  }
  if (!buf_page_make_young_if_needed(&block->page)) {
   buf_read_ahead_linear(id);
  }

  page = buf_block_get_frame(block);

  blob_header = page + offset;
  part_len = btr_blob_get_part_len(blob_header);
  copy_len = ut_min(part_len, len - copied_len);

  memcpy(buf + copied_len,
         blob_header + BTR_BLOB_HDR_SIZE, copy_len);
  copied_len += copy_len;

  id.set_page_no(btr_blob_get_next_page_no(blob_header));

  mtr_commit(&mtr);

  if (id.page_no() == FIL_NULL || copy_len != part_len) {
   MEM_CHECK_DEFINED(buf, copied_len);
   return(copied_len);
  }

  /* On other BLOB pages except the first the BLOB header
always is at the page data start: */

  offset = FIL_PAGE_DATA;

  ut_ad(copied_len <= len);
 }
}

/** Copies the prefix of a compressed BLOB.
The clustered index record that points to this BLOB must be protected
by a lock or a page latch.
@param[out] buf  the externally stored part of the field,
or a prefix of it
@param[in] len  length of buf, in bytes
@param[in] zip_size ROW_FORMAT=COMPRESSED page size
@param[in] id  page identifier of the BLOB pages
@return number of bytes written to buf */
static
ulint
btr_copy_zblob_prefix(
 byte*   buf,
 uint32_t  len,
 ulint   zip_size,
 page_id_t  id,
 uint32_t  offset)
{
 ulint  page_type = FIL_PAGE_TYPE_ZBLOB;
 mem_heap_t* heap;
 int  err;
 z_stream d_stream;

 d_stream.next_out = buf;
 d_stream.avail_out = static_cast<uInt>(len);
 d_stream.next_in = Z_NULL;
 d_stream.avail_in = 0;

 /* Zlib inflate needs 32 kilobytes for the default
window size, plus a few kilobytes for small objects. */
 heap = mem_heap_create(40000);
 page_zip_set_alloc(&d_stream, heap);

 ut_ad(zip_size);
 ut_ad(ut_is_2pow(zip_size));
 ut_ad(id.space());

 err = inflateInit(&d_stream);
 ut_a(err == Z_OK);

 for (;;) {
  buf_page_t* bpage;
  uint32_t next_page_no;

  bpage = buf_page_get_zip(id);

  if (UNIV_UNLIKELY(!bpage)) {
   goto func_exit;
  }

  if (UNIV_UNLIKELY
      (fil_page_get_type(bpage->zip.data) != page_type)) {

   ib::error() << "Unexpected type "
    << fil_page_get_type(bpage->zip.data)
    << " of compressed BLOB page " << id;

   ut_ad(0);
   goto end_of_blob;
  }

  next_page_no = mach_read_from_4(bpage->zip.data + offset);

  if (UNIV_LIKELY(offset == FIL_PAGE_NEXT)) {
   /* When the BLOB begins at page header,
   the compressed data payload does not
immediately follow the next page pointer. */
   offset = FIL_PAGE_DATA;
  } else {
   offset += 4;
  }

  d_stream.next_in = bpage->zip.data + offset;
  d_stream.avail_in = uInt(zip_size - offset);

  err = inflate(&d_stream, Z_NO_FLUSH);
  switch (err) {
  case Z_OK:
   if (!d_stream.avail_out) {
    goto end_of_blob;
   }
   break;
  case Z_STREAM_END:
   if (next_page_no == FIL_NULL) {
    goto end_of_blob;
   }
   /* fall through */
  default:
inflate_error:
   ib::error() << "inflate() of compressed BLOB page "
    << id
    << " returned " << err
    << " (" << d_stream.msg << ")";

  case Z_BUF_ERROR:
   goto end_of_blob;
  }

  if (next_page_no == FIL_NULL) {
   if (!d_stream.avail_in) {
    ib::error()
     << "Unexpected end of compressed "
     << "BLOB page " << id;
   } else {
    err = inflate(&d_stream, Z_FINISH);
    switch (err) {
    case Z_STREAM_END:
    case Z_BUF_ERROR:
     break;
    default:
     goto inflate_error;
    }
   }

end_of_blob:
   bpage->lock.s_unlock();
   goto func_exit;
  }

  bpage->lock.s_unlock();

  /* On other BLOB pages except the first
the BLOB header always is at the page header: */

  id.set_page_no(next_page_no);
  offset = FIL_PAGE_NEXT;
  page_type = FIL_PAGE_TYPE_ZBLOB2;
 }

func_exit:
 inflateEnd(&d_stream);
 mem_heap_free(heap);
 MEM_CHECK_DEFINED(buf, d_stream.total_out);
 return(d_stream.total_out);
}

/** Copies the prefix of an externally stored field of a record.
The clustered index record that points to this BLOB must be protected
by a lock or a page latch.
@param[out] buf  the externally stored part of the
field, or a prefix of it
@param[in] len  length of buf, in bytes
@param[in] zip_size ROW_FORMAT=COMPRESSED page size, or 0
@param[in] id  page identifier of the first BLOB page
@param[in] offset  offset on the first BLOB page
@return number of bytes written to buf */
static
ulint
btr_copy_externally_stored_field_prefix_low(
 byte*   buf,
 uint32_t  len,
 ulint   zip_size,
 page_id_t  id,
 uint32_t  offset)
{
  if (len == 0)
    return 0;

  return zip_size
    ? btr_copy_zblob_prefix(buf, len, zip_size, id, offset)
    : btr_copy_blob_prefix(buf, len, id, offset);
}

/** Copies the prefix of an externally stored field of a record.
The clustered index record must be protected by a lock or a page latch.
@param[out] buf  the field, or a prefix of it
@param[in] len  length of buf, in bytes
@param[in] zip_size ROW_FORMAT=COMPRESSED page size, or 0
@param[in] data  'internally' stored part of the field
containing also the reference to the external part; must be protected by
a lock or a page latch
@param[in] local_len length of data, in bytes
@return the length of the copied field, or 0 if the column was being
or has been deleted */
ulint
btr_copy_externally_stored_field_prefix(
 byte*   buf,
 ulint   len,
 ulint   zip_size,
 const byte*  data,
 ulint   local_len)
{
 ut_a(local_len >= BTR_EXTERN_FIELD_REF_SIZE);

 local_len -= BTR_EXTERN_FIELD_REF_SIZE;

 if (UNIV_UNLIKELY(local_len >= len)) {
  memcpy(buf, data, len);
  return(len);
 }

 memcpy(buf, data, local_len);
 data += local_len;

 ut_a(memcmp(data, field_ref_zero, BTR_EXTERN_FIELD_REF_SIZE));

 if (!mach_read_from_4(data + BTR_EXTERN_LEN + 4)) {
  /* The externally stored part of the column has been
  (partially) deleted.  Signal the half-deleted BLOB
to the caller. */

  return(0);
 }

 uint32_t space_id = mach_read_from_4(data + BTR_EXTERN_SPACE_ID);
 uint32_t page_no = mach_read_from_4(data + BTR_EXTERN_PAGE_NO);
 uint32_t offset = mach_read_from_4(data + BTR_EXTERN_OFFSET);
 len -= local_len;

 return(local_len
        + btr_copy_externally_stored_field_prefix_low(buf + local_len,
            uint32_t(len),
            zip_size,
            page_id_t(
             space_id,
             page_no),
            offset));
}

/** Copies an externally stored field of a record to mem heap.
The clustered index record must be protected by a lock or a page latch.
@param[out] len  length of the whole field
@param[in] data  'internally' stored part of the field
containing also the reference to the external part; must be protected by
a lock or a page latch
@param[in] zip_size ROW_FORMAT=COMPRESSED page size, or 0
@param[in] local_len length of data
@param[in,out] heap  mem heap
@return the whole field copied to heap */
byte*
btr_copy_externally_stored_field(
 ulint*   len,
 const byte*  data,
 ulint   zip_size,
 ulint   local_len,
 mem_heap_t*  heap)
{
 byte* buf;

 ut_a(local_len >= BTR_EXTERN_FIELD_REF_SIZE);

 local_len -= BTR_EXTERN_FIELD_REF_SIZE;

 uint32_t space_id = mach_read_from_4(data + local_len
          + BTR_EXTERN_SPACE_ID);
 uint32_t page_no = mach_read_from_4(data + local_len
         + BTR_EXTERN_PAGE_NO);
 uint32_t offset = mach_read_from_4(data + local_len
        + BTR_EXTERN_OFFSET);

 /* Currently a BLOB cannot be bigger than 4 GB; we
leave the 4 upper bytes in the length field unused */

 uint32_t extern_len = mach_read_from_4(data + local_len
            + BTR_EXTERN_LEN + 4);

 buf = (byte*) mem_heap_alloc(heap, local_len + extern_len);

 memcpy(buf, data, local_len);
 *len = local_len
  + btr_copy_externally_stored_field_prefix_low(buf + local_len,
             extern_len,
             zip_size,
             page_id_t(
              space_id,
              page_no),
             offset);

 return(buf);
}

/** Copies an externally stored field of a record to mem heap.
@param[in] rec  record in a clustered index; must be
protected by a lock or a page latch
@param[in] offset  array returned by rec_get_offsets()
@param[in] zip_size ROW_FORMAT=COMPRESSED page size, or 0
@param[in] no  field number
@param[out] len  length of the field
@param[in,out] heap  mem heap
@return the field copied to heap, or NULL if the field is incomplete */
byte*
btr_rec_copy_externally_stored_field(
 const rec_t*  rec,
 const rec_offs*  offsets,
 ulint   zip_size,
 ulint   no,
 ulint*   len,
 mem_heap_t*  heap)
{
 ulint  local_len;
 const byte* data;

 ut_a(rec_offs_nth_extern(offsets, no));

 /* An externally stored field can contain some initial
 data from the field, and in the last 20 bytes it has the
 space id, page number, and offset where the rest of the
 field data is stored, and the data length in addition to
 the data stored locally. We may need to store some data
 locally to get the local record length above the 128 byte
 limit so that field offsets are stored in two bytes, and
the extern bit is available in those two bytes. */

 data = rec_get_nth_field(rec, offsets, no, &local_len);

 ut_a(local_len >= BTR_EXTERN_FIELD_REF_SIZE);

 if (UNIV_UNLIKELY
     (!memcmp(data + local_len - BTR_EXTERN_FIELD_REF_SIZE,
       field_ref_zero, BTR_EXTERN_FIELD_REF_SIZE))) {
  /* The externally stored field was not written yet.
  This record should only be seen by
  trx_rollback_recovered() or any
TRX_ISO_READ_UNCOMMITTED transactions. */
  return(NULL);
 }

 return(btr_copy_externally_stored_field(len, data,
      zip_size, local_len, heap));
}