Impressum row0log.cc   Sprache: C

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

Copyright (c) 2011, 2018, Oracle and/or its affiliates. All Rights Reserved.
Copyright (c) 2017, 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 row/row0log.cc
Modification log for online index creation and online table rebuild

Created 2011-05-26 Marko Makela
*******************************************************/

#include "row0log.h"
#include "row0row.h"
#include "row0ins.h"
#include "row0upd.h"
#include "row0merge.h"
#include "row0ext.h"
#include "log0crypt.h"
#include "data0data.h"
#include "que0que.h"
#include "srv0mon.h"
#include "handler0alter.h"
#include "ut0stage.h"
#include "trx0rec.h"

#include <sql_class.h>
#include <algorithm>
#include <map>

Atomic_counter<ulint> onlineddl_rowlog_rows;
ulint onlineddl_rowlog_pct_used;
ulint onlineddl_pct_progress;

/** Table row modification operations during online table rebuild.
Delete-marked records are not copied to the rebuilt table. */
enum row_tab_op {
 /** Insert a record */
 ROW_T_INSERT = 0x41,
 /** Update a record in place */
 ROW_T_UPDATE,
 /** Delete (purge) a record */
 ROW_T_DELETE
};

/** Index record modification operations during online index creation */
enum row_op {
 /** Insert a record */
 ROW_OP_INSERT = 0x61,
 /** Delete a record */
 ROW_OP_DELETE
};

/** Size of the modification log entry header, in bytes */
#define ROW_LOG_HEADER_SIZE 2/*op, extra_size*/

/** Log block for modifications during online ALTER TABLE */
struct row_log_buf_t {
 byte*  block; /*!< file block buffer */
 size_t  size; /*!< length of block in bytes */
 ut_new_pfx_t block_pfx; /*!< opaque descriptor of "block". Set
    by ut_allocator::allocate_large() and fed to
ut_allocator::deallocate_large(). */
 mrec_buf_t buf; /*!< buffer for accessing a record
that spans two blocks */
 ulint  blocks; /*!< current position in blocks */
 ulint  bytes; /*!< current position within block */
 ulonglong total; /*!< logical position, in bytes from
    the start of the row_log_table log;
    0 for row_log_online_op() and
row_log_apply(). */
};

/** @brief Buffer for logging modifications during online index creation

All modifications to an index that is being created will be logged by
row_log_online_op() to this buffer.

All modifications to a table that is being rebuilt will be logged by
row_log_table_delete(), row_log_table_update(), row_log_table_insert()
to this buffer.

When head.blocks == tail.blocks, the reader will access tail.block
directly. When also head.bytes == tail.bytes, both counts will be
reset to 0 and the file will be truncated. */
struct row_log_t {
 pfs_os_file_t fd; /*!< file descriptor */
 mysql_mutex_t mutex; /*!< mutex protecting error,
max_trx and tail */
 dict_table_t* table; /*!< table that is being rebuilt,
    or NULL when this is a secondary
index that is being created online */
 bool  same_pk;/*!< whether the definition of the PRIMARY KEY
has remained the same */
 const dtuple_t* defaults;
    /*!< default values of added, changed columns,
or NULL */
 const ulint* col_map;/*!< mapping of old column numbers to
new ones, or NULL if !table */
 dberr_t  error; /*!< error that occurred during online
table rebuild */
 /** The transaction ID of the ALTER TABLE transaction.  Any
 concurrent DML would necessarily be logged with a larger
 transaction ID, because ha_innobase::prepare_inplace_alter_table()
 acts as a barrier that ensures that any concurrent transaction
 that operates on the table would have been started after
 ha_innobase::prepare_inplace_alter_table() returns and before
 ha_innobase::commit_inplace_alter_table(commit=true) is invoked.

 Due to the nondeterministic nature of purge and due to the
 possibility of upgrading from an earlier version of MariaDB
 or MySQL, it is possible that row_log_table_low() would be
 fed DB_TRX_ID that precedes than min_trx. We must normalize
such references to reset_trx_id[]. */
 trx_id_t min_trx;
 trx_id_t max_trx;/*!< biggest observed trx_id in
    row_log_online_op();
    protected by mutex and index->lock S-latch,
or by index->lock X-latch only */
 row_log_buf_t tail; /*!< writer context;
    protected by mutex and index->lock S-latch,
or by index->lock X-latch only */
 size_t  crypt_tail_size; /*!< size of crypt_tail_size*/
 byte*  crypt_tail; /*!< writer context;
    temporary buffer used in encryption,
decryption or NULL*/
 row_log_buf_t head; /*!< reader context; protected by MDL only;
modifiable by row_log_apply_ops() */
 size_t  crypt_head_size; /*!< size of crypt_tail_size*/
 byte*  crypt_head; /*!< reader context;
    temporary buffer used in encryption,
decryption or NULL */
 const char* path; /*!< where to create temporary file during
log operation */
 /** the number of core fields in the clustered index of the
 source table; before row_log_table_apply() completes, the
 table could be emptied, so that table->is_instant() no longer holds,
but all log records must be in the "instant" format. */
 unsigned n_core_fields;
 /** the default values of non-core fields when the operation started */
 dict_col_t::def_t* non_core_fields;
 bool  allow_not_null; /*!< Whether the alter ignore is being
    used or if the sql mode is non-strict mode;
    if not, NULL values will not be converted to
defaults */
 const TABLE* old_table; /*< Use old table in case of error. */

 uint64_t n_rows; /*< Number of rows read from the table */

 /** Alter table transaction. It can be used to apply the DML logs
into the table */
 const trx_t* alter_trx;

 /** Determine whether the log should be in the 'instant ADD' format
 @param[in] index the clustered index of the source table
@return whether to use the 'instant ADD COLUMN' format */
 bool is_instant(const dict_index_t* index) const
 {
  ut_ad(table);
  ut_ad(n_core_fields <= index->n_fields);
  return n_core_fields != index->n_fields;
 }

 const byte* instant_field_value(ulint n, ulint* len) const
 {
  ut_ad(n >= n_core_fields);
  const dict_col_t::def_t& d= non_core_fields[n - n_core_fields];
  *len = d.len;
  return static_cast<const byte*>(d.data);
 }
};

/** Create the file or online log if it does not exist.
@param[in,out] log     online rebuild log
@return true if success, false if not */
static MY_ATTRIBUTE((warn_unused_result))
pfs_os_file_t
row_log_tmpfile(
 row_log_t* log)
{
 DBUG_ENTER("row_log_tmpfile");
 if (log->fd == OS_FILE_CLOSED) {
  log->fd = row_merge_file_create_low(log->path);
  DBUG_EXECUTE_IF("row_log_tmpfile_fail",
    if (log->fd != OS_FILE_CLOSED)
     row_merge_file_destroy_low(log->fd);
    log->fd = OS_FILE_CLOSED;);
  if (log->fd != OS_FILE_CLOSED) {
   MONITOR_ATOMIC_INC(MONITOR_ALTER_TABLE_LOG_FILES);
  }
 }

 DBUG_RETURN(log->fd);
}

/** Allocate the memory for the log buffer.
@param[in,out] log_buf Buffer used for log operation
@return TRUE if success, false if not */
static MY_ATTRIBUTE((warn_unused_result))
bool
row_log_block_allocate(
 row_log_buf_t& log_buf)
{
 DBUG_ENTER("row_log_block_allocate");
 if (log_buf.block == NULL) {
  DBUG_EXECUTE_IF(
   "simulate_row_log_allocation_failure",
   DBUG_RETURN(false);
  );

  log_buf.block = ut_allocator<byte>(mem_key_row_log_buf)
   .allocate_large(srv_sort_buf_size,
     &log_buf.block_pfx);

  if (log_buf.block == NULL) {
   DBUG_RETURN(false);
  }
  log_buf.size = srv_sort_buf_size;
 }
 DBUG_RETURN(true);
}

/** Free the log buffer.
@param[in,out] log_buf Buffer used for log operation */
static
void
row_log_block_free(
 row_log_buf_t& log_buf)
{
 DBUG_ENTER("row_log_block_free");
 if (log_buf.block != NULL) {
  ut_allocator<byte>(mem_key_row_log_buf).deallocate_large(
   log_buf.block, &log_buf.block_pfx);
  log_buf.block = NULL;
 }
 DBUG_VOID_RETURN;
}

/** Logs an operation to a secondary index that is (or was) being created.
@param  index   index, S or X latched
@param  tuple   index tuple
@param  trx     transaction
@param  trx_id  transaction ID for insert, or 0 for delete
@retval false if row_log_apply() failure happens
or true otherwise */
static bool row_log_online_op(dict_index_t *index, const dtuple_t *tuple,
                              trx_t *trx, trx_id_t trx_id)
{
 byte*  b;
 ulint  extra_size;
 ulint  size;
 ulint  mrec_size;
 ulint  avail_size;
 row_log_t* log;
 bool  success= true;

 ut_ad(dtuple_validate(tuple));
 ut_ad(dtuple_get_n_fields(tuple) == dict_index_get_n_fields(index));
 ut_ad(index->lock.have_x() || index->lock.have_s());

 if (index->is_corrupted()) {
  return success;
 }

 ut_ad(dict_index_is_online_ddl(index)
       || (index->online_log
    && index->online_status == ONLINE_INDEX_COMPLETE));

 /* Compute the size of the record. This differs from
 row_merge_buf_encode(), because here we do not encode
extra_size+1 (and reserve 0 as the end-of-chunk marker). */

 size = rec_get_converted_size_temp<false>(
  index, tuple->fields, tuple->n_fields, &extra_size);
 ut_ad(size >= extra_size);
 ut_ad(size <= sizeof log->tail.buf);

 mrec_size = ROW_LOG_HEADER_SIZE
  + (extra_size >= 0x80) + size
  + (trx_id ? DATA_TRX_ID_LEN : 0);

 log = index->online_log;
 mysql_mutex_lock(&log->mutex);

start_log:
 if (trx_id > log->max_trx) {
  log->max_trx = trx_id;
 }

 if (!row_log_block_allocate(log->tail)) {
  log->error = DB_OUT_OF_MEMORY;
  goto err_exit;
 }

 MEM_UNDEFINED(log->tail.buf, sizeof log->tail.buf);

 ut_ad(log->tail.bytes < srv_sort_buf_size);
 avail_size = srv_sort_buf_size - log->tail.bytes;

 if (mrec_size > avail_size) {
  b = log->tail.buf;
 } else {
  b = log->tail.block + log->tail.bytes;
 }

 if (trx_id != 0) {
  *b++ = ROW_OP_INSERT;
  trx_write_trx_id(b, trx_id);
  b += DATA_TRX_ID_LEN;
 } else {
  *b++ = ROW_OP_DELETE;
 }

 if (extra_size < 0x80) {
  *b++ = (byte) extra_size;
 } else {
  ut_ad(extra_size < 0x8000);
  *b++ = (byte) (0x80 | (extra_size >> 8));
  *b++ = (byte) extra_size;
 }

 rec_convert_dtuple_to_temp<false>(
  b + extra_size, index, tuple->fields, tuple->n_fields);

 b += size;

 if (mrec_size >= avail_size) {
  const os_offset_t byte_offset
   = (os_offset_t) log->tail.blocks
   * srv_sort_buf_size;
  byte*   buf = log->tail.block;

  if (byte_offset + srv_sort_buf_size >= srv_online_max_size) {
   if (index->online_status != ONLINE_INDEX_COMPLETE)
    goto write_failed;
   /* About to run out of log, InnoDB has to
apply the online log for the completed index */
   index->lock.s_unlock();
   dberr_t error= row_log_apply(
    trx, index, nullptr, nullptr);
   index->lock.s_lock(SRW_LOCK_CALL);
   if (error != DB_SUCCESS) {
    /* Mark all newly added indexes
as corrupted */
    log->error = error;
    success = false;
    goto err_exit;
   }

   /* Recheck whether the index online log */
   if (!index->online_log) {
    goto err_exit;
   }

   goto start_log;
  }

  if (mrec_size == avail_size) {
   ut_ad(b == &buf[srv_sort_buf_size]);
  } else {
   ut_ad(b == log->tail.buf + mrec_size);
   memcpy(buf + log->tail.bytes,
          log->tail.buf, avail_size);
  }

  MEM_CHECK_DEFINED(buf, srv_sort_buf_size);

  if (row_log_tmpfile(log) == OS_FILE_CLOSED) {
   log->error = DB_OUT_OF_MEMORY;
   goto err_exit;
  }

  /* If encryption is enabled encrypt buffer before writing it
to file system. */
  if (srv_encrypt_log) {
   if (!log_tmp_block_encrypt(
        buf, srv_sort_buf_size,
        log->crypt_tail, byte_offset)) {
    log->error = DB_DECRYPTION_FAILED;
    goto write_failed;
   }

   srv_stats.n_rowlog_blocks_encrypted.inc();
   buf = log->crypt_tail;
  }

  log->tail.blocks++;
  DBUG_EXECUTE_IF("os_file_write_fail",
    log->error = DB_TEMP_FILE_WRITE_FAIL;
    goto write_failed;);

  if (os_file_write(
       IORequestWrite,
       "(modification log)",
       log->fd,
       buf, byte_offset, srv_sort_buf_size)
      != DB_SUCCESS) {
   log->error = DB_TEMP_FILE_WRITE_FAIL;
write_failed:
   index->type |= DICT_CORRUPT;
  }

  MEM_UNDEFINED(log->tail.block, srv_sort_buf_size);
  MEM_UNDEFINED(buf, srv_sort_buf_size);

  memcpy(log->tail.block, log->tail.buf + avail_size,
         mrec_size - avail_size);
  log->tail.bytes = mrec_size - avail_size;
 } else {
  log->tail.bytes += mrec_size;
  ut_ad(b == log->tail.block + log->tail.bytes);
 }

 MEM_UNDEFINED(log->tail.buf, sizeof log->tail.buf);
err_exit:
 mysql_mutex_unlock(&log->mutex);
 return success;
}

/******************************************************//**
Gets the error status of the online index rebuild log.
@return DB_SUCCESS or error code */
dberr_t
row_log_table_get_error(
/*====================*/
 const dict_index_t* index) /*!< in: clustered index of a table
that is being rebuilt online */
{
 ut_ad(dict_index_is_clust(index));
 ut_ad(dict_index_is_online_ddl(index));
 return(index->online_log->error);
}

/******************************************************//**
Starts logging an operation to a table that is being rebuilt.
@return pointer to log, or NULL if no logging is necessary */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
byte*
row_log_table_open(
/*===============*/
 row_log_t* log, /*!< in/out: online rebuild log */
 ulint  size, /*!< in: size of log record */
 ulint*  avail) /*!< out: available size for log record */
{
 mysql_mutex_lock(&log->mutex);

 MEM_UNDEFINED(log->tail.buf, sizeof log->tail.buf);

 if (log->error != DB_SUCCESS) {
err_exit:
  mysql_mutex_unlock(&log->mutex);
  return(NULL);
 }

 if (!row_log_block_allocate(log->tail)) {
  log->error = DB_OUT_OF_MEMORY;
  goto err_exit;
 }

 ut_ad(log->tail.bytes < srv_sort_buf_size);
 *avail = srv_sort_buf_size - log->tail.bytes;

 if (size > *avail) {
  /* Make sure log->tail.buf is large enough */
  ut_ad(size <= sizeof log->tail.buf);
  return(log->tail.buf);
 } else {
  return(log->tail.block + log->tail.bytes);
 }
}

/******************************************************//**
Stops logging an operation to a table that is being rebuilt. */
static MY_ATTRIBUTE((nonnull))
void
row_log_table_close_func(
/*=====================*/
 dict_index_t* index, /*!< in/out: online rebuilt index */
#ifdef UNIV_DEBUG
 const byte* b, /*!< in: end of log record */
#endif /* UNIV_DEBUG */
 ulint  size, /*!< in: size of log record */
 ulint  avail) /*!< in: available size for log record */
{
 row_log_t* log = index->online_log;

 mysql_mutex_assert_owner(&log->mutex);

 if (size >= avail) {
  const os_offset_t byte_offset
   = (os_offset_t) log->tail.blocks
   * srv_sort_buf_size;
  byte*   buf = log->tail.block;

  if (byte_offset + srv_sort_buf_size >= srv_online_max_size) {
   goto write_failed;
  }

  if (size == avail) {
   ut_ad(b == &buf[srv_sort_buf_size]);
  } else {
   ut_ad(b == log->tail.buf + size);
   memcpy(buf + log->tail.bytes, log->tail.buf, avail);
  }

  MEM_CHECK_DEFINED(buf, srv_sort_buf_size);

  if (row_log_tmpfile(log) == OS_FILE_CLOSED) {
   log->error = DB_OUT_OF_MEMORY;
   goto err_exit;
  }

  /* If encryption is enabled encrypt buffer before writing it
to file system. */
  if (srv_encrypt_log) {
   if (!log_tmp_block_encrypt(
        log->tail.block, srv_sort_buf_size,
        log->crypt_tail, byte_offset,
        index->table->space_id)) {
    log->error = DB_DECRYPTION_FAILED;
    goto err_exit;
   }

   srv_stats.n_rowlog_blocks_encrypted.inc();
   buf = log->crypt_tail;
  }

  log->tail.blocks++;
  if (os_file_write(
       IORequestWrite,
       "(modification log)",
       log->fd,
       buf, byte_offset, srv_sort_buf_size)
      != DB_SUCCESS) {
write_failed:
   log->error = DB_ONLINE_LOG_TOO_BIG;
  }

  MEM_UNDEFINED(log->tail.block, srv_sort_buf_size);
  MEM_UNDEFINED(buf, srv_sort_buf_size);
  memcpy(log->tail.block, log->tail.buf + avail, size - avail);
  log->tail.bytes = size - avail;
 } else {
  log->tail.bytes += size;
  ut_ad(b == log->tail.block + log->tail.bytes);
 }

 log->tail.total += size;
 MEM_UNDEFINED(log->tail.buf, sizeof log->tail.buf);
err_exit:
 mysql_mutex_unlock(&log->mutex);

 onlineddl_rowlog_rows++;
 /* 10000 means 100.00%, 4525 means 45.25% */
 onlineddl_rowlog_pct_used = static_cast<ulint>((log->tail.total * 10000) / srv_online_max_size);
}

#ifdef UNIV_DEBUG
# define row_log_table_close(index, b, size, avail) \
 row_log_table_close_func(index, b, size, avail)
#else /* UNIV_DEBUG */
# define row_log_table_close(log, b, size, avail) \
 row_log_table_close_func(index, size, avail)
#endif /* UNIV_DEBUG */

ATTRIBUTE_COLD
void row_log_mark_virtual_cols(const dict_table_t *table,
                               TABLE *maria_table) noexcept
{
  const row_log_t *const log= dict_table_get_first_index(table)->online_log;
  ut_ad(log);
  const dict_v_col_t *v_cols= table->v_cols, *old_v_cols= log->table->v_cols;
  ut_d(const unsigned n_v_cols= log->table->n_v_cols);
  ut_ad(n_v_cols == table->n_v_cols);
  for (uint j= maria_table->s->virtual_fields, num_v= 0; j--; )
  {
    Field *vf= maria_table->vfield[j];
    if (vf->stored_in_db())
      continue;
    ut_ad(num_v < n_v_cols);
    ut_ad(num_v < table->n_v_cols);
    if (v_cols[num_v].m_col.ord_part | old_v_cols[num_v].m_col.ord_part)
      maria_table->mark_virtual_column_with_deps(vf);
    num_v++;
  }
}

/******************************************************//**
Logs a delete operation to a table that is being rebuilt.
This will be merged in row_log_table_apply_delete(). */
void
row_log_table_delete(
/*=================*/
 const rec_t* rec, /*!< in: clustered index leaf page record,
page X-latched */
 dict_index_t* index, /*!< in/out: clustered index, S-latched
or X-latched */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec,index) */
 const byte* sys) /*!< in: DB_TRX_ID,DB_ROLL_PTR that should
be logged, or NULL to use those in rec */
{
 ulint  old_pk_extra_size;
 ulint  old_pk_size;
 ulint  mrec_size;
 ulint  avail_size;
 mem_heap_t* heap  = NULL;
 const dtuple_t* old_pk;

 ut_ad(dict_index_is_clust(index));
 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(rec_offs_n_fields(offsets) == dict_index_get_n_fields(index));
 ut_ad(rec_offs_size(offsets) <= sizeof index->online_log->tail.buf);
 ut_ad(index->lock.have_any());

 if (index->online_status != ONLINE_INDEX_CREATION
     || (index->type & DICT_CORRUPT) || index->table->corrupted
     || index->online_log->error != DB_SUCCESS) {
  return;
 }

 dict_table_t* new_table = index->online_log->table;
 dict_index_t* new_index = dict_table_get_first_index(new_table);

 ut_ad(dict_index_is_clust(new_index));
 ut_ad(!dict_index_is_online_ddl(new_index));
 ut_ad(index->online_log->min_trx);

 /* Create the tuple PRIMARY KEY,DB_TRX_ID,DB_ROLL_PTR in new_table. */
 if (index->online_log->same_pk) {
  dtuple_t* tuple;
  ut_ad(new_index->n_uniq == index->n_uniq);

  /* The PRIMARY KEY and DB_TRX_ID,DB_ROLL_PTR are in the first
fields of the record. */
  heap = mem_heap_create(
   DATA_TRX_ID_LEN
   + DTUPLE_EST_ALLOC(new_index->first_user_field()));
  old_pk = tuple = dtuple_create(heap,
            new_index->first_user_field());
  dict_index_copy_types(tuple, new_index, tuple->n_fields);
  dtuple_set_n_fields_cmp(tuple, new_index->n_uniq);

  for (ulint i = 0; i < dtuple_get_n_fields(tuple); i++) {
   ulint  len;
   const void* field = rec_get_nth_field(
    rec, offsets, i, &len);
   dfield_t* dfield = dtuple_get_nth_field(
    tuple, i);
   ut_ad(len != UNIV_SQL_NULL);
   ut_ad(!rec_offs_nth_extern(offsets, i));
   dfield_set_data(dfield, field, len);
  }

  dfield_t* db_trx_id = dtuple_get_nth_field(
   tuple, new_index->n_uniq);

  const bool replace_sys_fields
   = sys
   || trx_read_trx_id(static_cast<byte*>(db_trx_id->data))
   < index->online_log->min_trx;

  if (replace_sys_fields) {
   if (!sys || trx_read_trx_id(sys)
       < index->online_log->min_trx) {
    sys = reset_trx_id;
   }

   dfield_set_data(db_trx_id, sys, DATA_TRX_ID_LEN);
   dfield_set_data(db_trx_id + 1, sys + DATA_TRX_ID_LEN,
     DATA_ROLL_PTR_LEN);
  }

  ut_d(trx_id_check(db_trx_id->data,
      index->online_log->min_trx));
 } else {
  /* The PRIMARY KEY has changed. Translate the tuple. */
  old_pk = row_log_table_get_pk(
   rec, index, offsets, NULL, &heap);

  if (!old_pk) {
   ut_ad(index->online_log->error != DB_SUCCESS);
   if (heap) {
    goto func_exit;
   }
   return;
  }
 }

 ut_ad(DATA_TRX_ID_LEN == dtuple_get_nth_field(
        old_pk, old_pk->n_fields - 2)->len);
 ut_ad(DATA_ROLL_PTR_LEN == dtuple_get_nth_field(
        old_pk, old_pk->n_fields - 1)->len);
 old_pk_size = rec_get_converted_size_temp<false>(
  new_index, old_pk->fields, old_pk->n_fields,
  &old_pk_extra_size);
 ut_ad(old_pk_extra_size < 0x100);

 /* 2 = 1 (extra_size) + at least 1 byte payload */
 mrec_size = 2 + old_pk_size;

 if (byte* b = row_log_table_open(index->online_log,
      mrec_size, &avail_size)) {
  *b++ = ROW_T_DELETE;
  *b++ = static_cast<byte>(old_pk_extra_size);

  rec_convert_dtuple_to_temp<false>(
   b + old_pk_extra_size, new_index,
   old_pk->fields, old_pk->n_fields);

  b += old_pk_size;

  row_log_table_close(index, b, mrec_size, avail_size);
 }

func_exit:
 mem_heap_free(heap);
}

/******************************************************//**
Logs an insert or update to a table that is being rebuilt. */
static
void
row_log_table_low_redundant(
/*========================*/
 const rec_t*  rec, /*!< in: clustered index leaf
     page record in ROW_FORMAT=REDUNDANT,
page X-latched */
 dict_index_t*  index, /*!< in/out: clustered index, S-latched
or X-latched */
 bool   insert, /*!< in: true if insert,
false if update */
 const dtuple_t*  old_pk, /*!< in: old PRIMARY KEY value
     (if !insert and a PRIMARY KEY
is being created) */
 const dict_index_t* new_index)
     /*!< in: clustered index of the
new table, not latched */
{
 ulint  old_pk_size;
 ulint  old_pk_extra_size;
 ulint  size;
 ulint  extra_size;
 ulint  mrec_size;
 ulint  avail_size;
 mem_heap_t* heap  = NULL;
 dtuple_t* tuple;
 const auto n_fields = rec_get_n_fields_old(rec);

 ut_ad(index->n_fields >= n_fields);
 ut_ad(index->n_fields == n_fields || index->is_instant());
 ut_ad(dict_tf2_is_valid(index->table->flags, index->table->flags2));
 ut_ad(!dict_table_is_comp(index->table));  /* redundant row format */
 ut_ad(dict_index_is_clust(new_index));

 heap = mem_heap_create(DTUPLE_EST_ALLOC(n_fields));
 tuple = dtuple_create(heap, n_fields);
 dict_index_copy_types(tuple, index, n_fields);

 dtuple_set_n_fields_cmp(tuple, dict_index_get_n_unique(index));

 if (rec_get_1byte_offs_flag(rec)) {
  for (ulint i = 0; i < n_fields; i++) {
   dfield_t* dfield;
   ulint  len;
   const void* field;

   dfield = dtuple_get_nth_field(tuple, i);
   field = rec_get_nth_field_old(rec, i, &len);

   dfield_set_data(dfield, field, len);
  }
 } else {
  for (ulint i = 0; i < n_fields; i++) {
   dfield_t* dfield;
   ulint  len;
   const void* field;

   dfield = dtuple_get_nth_field(tuple, i);
   field = rec_get_nth_field_old(rec, i, &len);

   dfield_set_data(dfield, field, len);

   if (rec_2_is_field_extern(rec, i)) {
    dfield_set_ext(dfield);
   }
  }
 }

 dfield_t* db_trx_id = dtuple_get_nth_field(tuple, index->n_uniq);
 ut_ad(dfield_get_len(db_trx_id) == DATA_TRX_ID_LEN);
 ut_ad(dfield_get_len(db_trx_id + 1) == DATA_ROLL_PTR_LEN);

 if (trx_read_trx_id(static_cast<const byte*>
       (dfield_get_data(db_trx_id)))
     < index->online_log->min_trx) {
  dfield_set_data(db_trx_id, reset_trx_id, DATA_TRX_ID_LEN);
  dfield_set_data(db_trx_id + 1, reset_trx_id + DATA_TRX_ID_LEN,
    DATA_ROLL_PTR_LEN);
 }

 const bool is_instant = index->online_log->is_instant(index);
 rec_comp_status_t status = is_instant
  ? REC_STATUS_INSTANT : REC_STATUS_ORDINARY;

 size = rec_get_converted_size_temp<true>(
  index, tuple->fields, tuple->n_fields, &extra_size, status);
 if (is_instant) {
  size++;
  extra_size++;
 }

 mrec_size = ROW_LOG_HEADER_SIZE + size + (extra_size >= 0x80);

 if (insert || index->online_log->same_pk) {
  ut_ad(!old_pk);
  old_pk_extra_size = old_pk_size = 0;
 } else {
  ut_ad(old_pk);
  ut_ad(old_pk->n_fields == 2 + old_pk->n_fields_cmp);
  ut_ad(DATA_TRX_ID_LEN == dtuple_get_nth_field(
         old_pk, old_pk->n_fields - 2)->len);
  ut_ad(DATA_ROLL_PTR_LEN == dtuple_get_nth_field(
         old_pk, old_pk->n_fields - 1)->len);

  old_pk_size = rec_get_converted_size_temp<false>(
   new_index, old_pk->fields, old_pk->n_fields,
   &old_pk_extra_size);
  ut_ad(old_pk_extra_size < 0x100);
  mrec_size += 1/*old_pk_extra_size*/ + old_pk_size;
 }

 if (byte* b = row_log_table_open(index->online_log,
      mrec_size, &avail_size)) {
  if (insert) {
   *b++ = ROW_T_INSERT;
  } else {
   *b++ = ROW_T_UPDATE;

   if (old_pk_size) {
    *b++ = static_cast<byte>(old_pk_extra_size);

    rec_convert_dtuple_to_temp<false>(
     b + old_pk_extra_size, new_index,
     old_pk->fields, old_pk->n_fields);
    b += old_pk_size;
   }
  }

  if (extra_size < 0x80) {
   *b++ = static_cast<byte>(extra_size);
  } else {
   ut_ad(extra_size < 0x8000);
   *b++ = static_cast<byte>(0x80 | (extra_size >> 8));
   *b++ = static_cast<byte>(extra_size);
  }

  if (status == REC_STATUS_INSTANT) {
   ut_ad(is_instant);
   if (n_fields <= index->online_log->n_core_fields) {
    status = REC_STATUS_ORDINARY;
   }
   *b = status;
  }

  rec_convert_dtuple_to_temp<true>(
   b + extra_size, index, tuple->fields, tuple->n_fields,
   status);
  b += size;

  row_log_table_close(index, b, mrec_size, avail_size);
 }

 mem_heap_free(heap);
}

/******************************************************//**
Logs an insert or update to a table that is being rebuilt. */
static
void
row_log_table_low(
/*==============*/
 const rec_t* rec, /*!< in: clustered index leaf page record,
page X-latched */
 dict_index_t* index, /*!< in/out: clustered index, S-latched
or X-latched */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec,index) */
 bool  insert, /*!< in: true if insert, false if update */
 const dtuple_t* old_pk) /*!< in: old PRIMARY KEY value (if !insert
and a PRIMARY KEY is being created) */
{
 ulint   old_pk_size;
 ulint   old_pk_extra_size;
 ulint   extra_size;
 ulint   mrec_size;
 ulint   avail_size;
 const dict_index_t* new_index;
 row_log_t*  log = index->online_log;

 new_index = dict_table_get_first_index(log->table);

 ut_ad(dict_index_is_clust(index));
 ut_ad(dict_index_is_clust(new_index));
 ut_ad(!dict_index_is_online_ddl(new_index));
 ut_ad(rec_offs_validate(rec, index, offsets));
 ut_ad(rec_offs_n_fields(offsets) == dict_index_get_n_fields(index));
 ut_ad(rec_offs_size(offsets) <= sizeof log->tail.buf);
 ut_ad(index->lock.have_any());

 /* old_pk=row_log_table_get_pk() [not needed in INSERT] is a prefix
 of the clustered index record (PRIMARY KEY,DB_TRX_ID,DB_ROLL_PTR),
with no information on virtual columns */
 ut_ad(!old_pk || !insert);
 ut_ad(!old_pk || old_pk->n_v_fields == 0);

 if (index->online_status != ONLINE_INDEX_CREATION
     || (index->type & DICT_CORRUPT) || index->table->corrupted
     || log->error != DB_SUCCESS) {
  return;
 }

 if (!rec_offs_comp(offsets)) {
  row_log_table_low_redundant(
   rec, index, insert, old_pk, new_index);
  return;
 }

 ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY
       || rec_get_status(rec) == REC_STATUS_INSTANT);

 const ulint omit_size = REC_N_NEW_EXTRA_BYTES;

 const ulint rec_extra_size = rec_offs_extra_size(offsets) - omit_size;
 const bool is_instant = log->is_instant(index);
 extra_size = rec_extra_size + is_instant;

 unsigned fake_extra_size = 0;
 byte fake_extra_buf[3];
 if (is_instant && UNIV_UNLIKELY(!index->is_instant())) {
  /* The source table was emptied after ALTER TABLE
  started, and it was converted to non-instant format.
  Because row_log_table_apply_op() expects to find
  all records to be logged in the same way, we will
  be unable to copy the rec_extra_size bytes from the
record header, but must convert them here. */
  unsigned n_add = index->n_fields - 1 - log->n_core_fields;
  fake_extra_size = rec_get_n_add_field_len(n_add);
  ut_ad(fake_extra_size == 1 || fake_extra_size == 2);
  extra_size += fake_extra_size;
  byte* fake_extra = fake_extra_buf + fake_extra_size;
  rec_set_n_add_field(fake_extra, n_add);
  ut_ad(fake_extra == fake_extra_buf);
 }

 mrec_size = ROW_LOG_HEADER_SIZE
  + (extra_size >= 0x80) + rec_offs_size(offsets) - omit_size
  + is_instant + fake_extra_size;

 if (insert || log->same_pk) {
  ut_ad(!old_pk);
  old_pk_extra_size = old_pk_size = 0;
 } else {
  ut_ad(old_pk);
  ut_ad(old_pk->n_fields == 2 + old_pk->n_fields_cmp);
  ut_ad(DATA_TRX_ID_LEN == dtuple_get_nth_field(
         old_pk, old_pk->n_fields - 2)->len);
  ut_ad(DATA_ROLL_PTR_LEN == dtuple_get_nth_field(
         old_pk, old_pk->n_fields - 1)->len);

  old_pk_size = rec_get_converted_size_temp<false>(
   new_index, old_pk->fields, old_pk->n_fields,
   &old_pk_extra_size);
  ut_ad(old_pk_extra_size < 0x100);
  mrec_size += 1/*old_pk_extra_size*/ + old_pk_size;
 }

 if (byte* b = row_log_table_open(log, mrec_size, &avail_size)) {
  if (insert) {
   *b++ = ROW_T_INSERT;
  } else {
   *b++ = ROW_T_UPDATE;

   if (old_pk_size) {
    *b++ = static_cast<byte>(old_pk_extra_size);

    rec_convert_dtuple_to_temp<false>(
     b + old_pk_extra_size, new_index,
     old_pk->fields, old_pk->n_fields);
    b += old_pk_size;
   }
  }

  if (extra_size < 0x80) {
   *b++ = static_cast<byte>(extra_size);
  } else {
   ut_ad(extra_size < 0x8000);
   *b++ = static_cast<byte>(0x80 | (extra_size >> 8));
   *b++ = static_cast<byte>(extra_size);
  }

  if (is_instant) {
   *b++ = fake_extra_size
    ? REC_STATUS_INSTANT
    : rec_get_status(rec);
  } else {
   ut_ad(rec_get_status(rec) == REC_STATUS_ORDINARY);
  }

  memcpy(b, rec - rec_extra_size - omit_size, rec_extra_size);
  b += rec_extra_size;
  memcpy(b, fake_extra_buf + 1, fake_extra_size);
  b += fake_extra_size;
  ulint len;
  ulint trx_id_offs = rec_get_nth_field_offs(
   offsets, index->n_uniq, &len);
  ut_ad(len == DATA_TRX_ID_LEN);
  memcpy(b, rec, rec_offs_data_size(offsets));
  if (trx_read_trx_id(b + trx_id_offs) < log->min_trx) {
   memcpy(b + trx_id_offs,
          reset_trx_id, sizeof reset_trx_id);
  }
  b += rec_offs_data_size(offsets);

  row_log_table_close(index, b, mrec_size, avail_size);
 }
}

/******************************************************//**
Logs an update to a table that is being rebuilt.
This will be merged in row_log_table_apply_update(). */
void
row_log_table_update(
/*=================*/
 const rec_t* rec, /*!< in: clustered index leaf page record,
page X-latched */
 dict_index_t* index, /*!< in/out: clustered index, S-latched
or X-latched */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec,index) */
 const dtuple_t* old_pk) /*!< in: row_log_table_get_pk()
before the update */
{
 row_log_table_low(rec, index, offsets, false, old_pk);
}

/** Gets the old table column of a PRIMARY KEY column.
@param table old table (before ALTER TABLE)
@param col_map mapping of old column numbers to new ones
@param col_no column position in the new table
@return old table column, or NULL if this is an added column */
static
const dict_col_t*
row_log_table_get_pk_old_col(
/*=========================*/
 const dict_table_t* table,
 const ulint*  col_map,
 ulint   col_no)
{
 for (ulint i = 0; i < table->n_cols; i++) {
  if (col_no == col_map[i]) {
   return(dict_table_get_nth_col(table, i));
  }
 }

 return(NULL);
}

/** Maps an old table column of a PRIMARY KEY column.
@param[in] ifield  clustered index field in the new table (after
ALTER TABLE)
@param[in] index  the clustered index of ifield
@param[in,out] dfield  clustered index tuple field in the new table
@param[in,out] heap  memory heap for allocating dfield contents
@param[in] rec  clustered index leaf page record in the old
table
@param[in] offsets  rec_get_offsets(rec)
@param[in] i  rec field corresponding to col
@param[in] zip_size ROW_FORMAT=COMPRESSED size of the old table
@param[in] max_len  maximum length of dfield
@param[in] log  row log for the table
@retval DB_INVALID_NULL  if a NULL value is encountered
@retval DB_TOO_BIG_INDEX_COL if the maximum prefix length is exceeded */
static
dberr_t
row_log_table_get_pk_col(
 const dict_field_t* ifield,
 const dict_index_t* index,
 dfield_t*  dfield,
 mem_heap_t*  heap,
 const rec_t*  rec,
 const rec_offs*  offsets,
 ulint   i,
 ulint   zip_size,
 ulint   max_len,
 const row_log_t* log)
{
 const byte* field;
 ulint  len;

 field = rec_get_nth_field(rec, offsets, i, &len);

 if (len == UNIV_SQL_DEFAULT) {
  field = log->instant_field_value(i, &len);
 }

 if (len == UNIV_SQL_NULL) {
  if (!log->allow_not_null) {
   return(DB_INVALID_NULL);
  }

  unsigned col_no= ifield->col->ind;
  ut_ad(col_no < log->defaults->n_fields);

  field = static_cast<const byte*>(
   log->defaults->fields[col_no].data);
  if (!field) {
   return(DB_INVALID_NULL);
  }
  len = log->defaults->fields[col_no].len;
 }

 if (rec_offs_nth_extern(offsets, i)) {
  ulint field_len = ifield->prefix_len;
  byte* blob_field;

  if (!field_len) {
   field_len = ifield->fixed_len;
   if (!field_len) {
    field_len = max_len + 1;
   }
  }

  blob_field = static_cast<byte*>(
   mem_heap_alloc(heap, field_len));

  len = btr_copy_externally_stored_field_prefix(
   blob_field, field_len, zip_size, field, len);
  if (len >= max_len + 1) {
   return(DB_TOO_BIG_INDEX_COL);
  }

  dfield_set_data(dfield, blob_field, len);
 } else {
  dfield_set_data(dfield, mem_heap_dup(heap, field, len), len);
 }

 return(DB_SUCCESS);
}

/******************************************************//**
Constructs the old PRIMARY KEY and DB_TRX_ID,DB_ROLL_PTR
of a table that is being rebuilt.
@return tuple of PRIMARY KEY,DB_TRX_ID,DB_ROLL_PTR in the rebuilt table,
or NULL if the PRIMARY KEY definition does not change */
const dtuple_t*
row_log_table_get_pk(
/*=================*/
 const rec_t* rec, /*!< in: clustered index leaf page record,
page X-latched */
 dict_index_t* index, /*!< in/out: clustered index, S-latched
or X-latched */
 const rec_offs* offsets,/*!< in: rec_get_offsets(rec,index) */
 byte*  sys, /*!< out: DB_TRX_ID,DB_ROLL_PTR for
row_log_table_delete(), or NULL */
 mem_heap_t** heap) /*!< in/out: memory heap where allocated */
{
 dtuple_t* tuple = NULL;
 row_log_t* log = index->online_log;

 ut_ad(dict_index_is_clust(index));
 ut_ad(dict_index_is_online_ddl(index));
 ut_ad(!offsets || rec_offs_validate(rec, index, offsets));
 ut_ad(index->lock.have_any());
 ut_ad(log);
 ut_ad(log->table);
 ut_ad(log->min_trx);

 if (log->same_pk) {
  /* The PRIMARY KEY columns are unchanged. */
  if (sys) {
   /* Store the DB_TRX_ID,DB_ROLL_PTR. */
   ulint trx_id_offs = index->trx_id_offset;

   if (!trx_id_offs) {
    ulint len;

    if (!offsets) {
     offsets = rec_get_offsets(
      rec, index, nullptr,
      index->n_core_fields,
      index->db_trx_id() + 1, heap);
    }

    trx_id_offs = rec_get_nth_field_offs(
     offsets, index->db_trx_id(), &len);
    ut_ad(len == DATA_TRX_ID_LEN);
   }

   const byte* ptr = trx_read_trx_id(rec + trx_id_offs)
    < log->min_trx
    ? reset_trx_id
    : rec + trx_id_offs;

   memcpy(sys, ptr, DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
   ut_d(trx_id_check(sys, log->min_trx));
  }

  return(NULL);
 }

 mysql_mutex_lock(&log->mutex);

 /* log->error is protected by log->mutex. */
 if (log->error == DB_SUCCESS) {
  dict_table_t* new_table = log->table;
  dict_index_t* new_index
   = dict_table_get_first_index(new_table);
  const ulint new_n_uniq
   = dict_index_get_n_unique(new_index);

  if (!*heap) {
   ulint size = 0;

   if (!offsets) {
    size += (1 + REC_OFFS_HEADER_SIZE
      + unsigned(index->n_fields))
     * sizeof *offsets;
   }

   for (ulint i = 0; i < new_n_uniq; i++) {
    size += dict_col_get_min_size(
     dict_index_get_nth_col(new_index, i));
   }

   *heap = mem_heap_create(
    DTUPLE_EST_ALLOC(new_n_uniq + 2) + size);
  }

  if (!offsets) {
   offsets = rec_get_offsets(rec, index, nullptr,
        index->n_core_fields,
        ULINT_UNDEFINED, heap);
  }

  tuple = dtuple_create(*heap, new_n_uniq + 2);
  dict_index_copy_types(tuple, new_index, tuple->n_fields);
  dtuple_set_n_fields_cmp(tuple, new_n_uniq);

  const ulint max_len = DICT_MAX_FIELD_LEN_BY_FORMAT(new_table);

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

  for (ulint new_i = 0; new_i < new_n_uniq; new_i++) {
   dict_field_t* ifield;
   dfield_t* dfield;
   ulint  prtype;
   ulint  mbminlen, mbmaxlen;

   ifield = dict_index_get_nth_field(new_index, new_i);
   dfield = dtuple_get_nth_field(tuple, new_i);

   const ulint col_no
    = dict_field_get_col(ifield)->ind;

   if (const dict_col_t* col
       = row_log_table_get_pk_old_col(
        index->table, log->col_map, col_no)) {
    ulint i = dict_col_get_clust_pos(col, index);

    if (i == ULINT_UNDEFINED) {
     ut_ad(0);
     log->error = DB_CORRUPTION;
     goto err_exit;
    }

    log->error = row_log_table_get_pk_col(
     ifield, new_index, dfield, *heap,
     rec, offsets, i, zip_size, max_len,
     log);

    if (log->error != DB_SUCCESS) {
err_exit:
     tuple = NULL;
     goto func_exit;
    }

    mbminlen = col->mbminlen;
    mbmaxlen = col->mbmaxlen;
    prtype = col->prtype;
   } else {
    /* No matching column was found in the old
    table, so this must be an added column.
Copy the default value. */
    ut_ad(log->defaults);

    dfield_copy(dfield, dtuple_get_nth_field(
          log->defaults, col_no));
    mbminlen = dfield->type.mbminlen;
    mbmaxlen = dfield->type.mbmaxlen;
    prtype = dfield->type.prtype;
   }

   ut_ad(!dfield_is_ext(dfield));
   ut_ad(!dfield_is_null(dfield));

   if (ifield->prefix_len) {
    ulint len = dtype_get_at_most_n_mbchars(
     prtype, mbminlen, mbmaxlen,
     ifield->prefix_len,
     dfield_get_len(dfield),
     static_cast<const char*>(
      dfield_get_data(dfield)));

    ut_ad(len <= dfield_get_len(dfield));
    dfield_set_len(dfield, len);
   }
  }

  const byte* trx_roll = rec
   + row_get_trx_id_offset(index, offsets);

  /* Copy the fields, because the fields will be updated
  or the record may be moved somewhere else in the B-tree
as part of the upcoming operation. */
  if (trx_read_trx_id(trx_roll) < log->min_trx) {
   trx_roll = reset_trx_id;
   if (sys) {
    memcpy(sys, trx_roll,
           DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
   }
  } else if (sys) {
   memcpy(sys, trx_roll,
          DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN);
   trx_roll = sys;
  } else {
   trx_roll = static_cast<const byte*>(
    mem_heap_dup(
     *heap, trx_roll,
     DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN));
  }

  ut_d(trx_id_check(trx_roll, log->min_trx));

  dfield_set_data(dtuple_get_nth_field(tuple, new_n_uniq),
    trx_roll, DATA_TRX_ID_LEN);
  dfield_set_data(dtuple_get_nth_field(tuple, new_n_uniq + 1),
    trx_roll + DATA_TRX_ID_LEN, DATA_ROLL_PTR_LEN);
 }

func_exit:
 mysql_mutex_unlock(&log->mutex);
 return(tuple);
}

/******************************************************//**
Logs an insert to a table that is being rebuilt.
This will be merged in row_log_table_apply_insert(). */
void
row_log_table_insert(
/*=================*/
 const rec_t* rec, /*!< in: clustered index leaf page record,
page X-latched */
 dict_index_t* index, /*!< in/out: clustered index, S-latched
or X-latched */
 const rec_offs* offsets)/*!< in: rec_get_offsets(rec,index) */
{
 row_log_table_low(rec, index, offsets, true, NULL);
}

/******************************************************//**
Converts a log record to a table row.
@return converted row, or NULL if the conversion fails */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
const dtuple_t*
row_log_table_apply_convert_mrec(
/*=============================*/
 const mrec_t*  mrec,  /*!< in: merge record */
 dict_index_t*  index,  /*!< in: index of mrec */
 const rec_offs*  offsets, /*!< in: offsets of mrec */
 row_log_t*  log,  /*!< in: rebuild context */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 dberr_t*  error)  /*!< out: DB_SUCCESS or
      DB_MISSING_HISTORY or
reason of failure */
{
 dtuple_t* row;

 log->n_rows++;
 *error = DB_SUCCESS;

 /* This is based on row_build(). */
 if (log->defaults) {
  row = dtuple_copy(log->defaults, heap);
  /* dict_table_copy_types() would set the fields to NULL */
  for (ulint i = 0; i < dict_table_get_n_cols(log->table); i++) {
   dict_col_copy_type(
    dict_table_get_nth_col(log->table, i),
    dfield_get_type(dtuple_get_nth_field(row, i)));
  }
 } else {
  row = dtuple_create(heap, dict_table_get_n_cols(log->table));
  dict_table_copy_types(row, log->table);
 }

 for (ulint i = 0; i < rec_offs_n_fields(offsets); i++) {
  const dict_field_t* ind_field
   = dict_index_get_nth_field(index, i);

  if (ind_field->prefix_len) {
   /* Column prefixes can only occur in key
   fields, which cannot be stored externally. For
   a column prefix, there should also be the full
   field in the clustered index tuple. The row
tuple comprises full fields, not prefixes. */
   ut_ad(!rec_offs_nth_extern(offsets, i));
   continue;
  }

  const dict_col_t* col
   = dict_field_get_col(ind_field);

  if (col->is_dropped()) {
   /* the column was instantly dropped earlier */
   ut_ad(index->table->instant);
   continue;
  }

  ulint   col_no
   = log->col_map[dict_col_get_no(col)];

  if (col_no == ULINT_UNDEFINED) {
   /* the column is being dropped now */
   continue;
  }

  dfield_t* dfield
   = dtuple_get_nth_field(row, col_no);

  ulint   len;
  const byte*  data;

  if (rec_offs_nth_extern(offsets, i)) {
   ut_ad(rec_offs_any_extern(offsets));
   index->lock.x_lock(SRW_LOCK_CALL);

   data = btr_rec_copy_externally_stored_field(
    mrec, offsets,
    index->table->space->zip_size(),
    i, &len, heap);
   ut_a(data);
   dfield_set_data(dfield, data, len);

   index->lock.x_unlock();
  } else {
   data = rec_get_nth_field(mrec, offsets, i, &len);
   if (len == UNIV_SQL_DEFAULT) {
    data = log->instant_field_value(i, &len);
   }
   dfield_set_data(dfield, data, len);
  }

  if (len != UNIV_SQL_NULL && col->mtype == DATA_MYSQL
      && col->len != len && !dict_table_is_comp(log->table)) {

   ut_ad(col->len >= len);
   if (dict_table_is_comp(index->table)) {
    byte* buf = (byte*) mem_heap_alloc(heap,
             col->len);
    memcpy(buf, dfield->data, len);
    memset(buf + len, 0x20, col->len - len);

    dfield_set_data(dfield, buf, col->len);
   } else {
    /* field length mismatch should not happen
    when rebuilding the redundant row format
table. */
    ut_ad(0);
    *error = DB_CORRUPTION;
    return(NULL);
   }
  }

  /* See if any columns were changed to NULL or NOT NULL. */
  const dict_col_t* new_col
   = dict_table_get_nth_col(log->table, col_no);
  ut_ad(new_col->same_format(*col));

  /* Assert that prtype matches except for nullability. */
  ut_ad(!((new_col->prtype ^ dfield_get_type(dfield)->prtype)
   & ~(DATA_NOT_NULL | DATA_VERSIONED
       | CHAR_COLL_MASK << 16 | DATA_LONG_TRUE_VARCHAR)));

  if (new_col->prtype == col->prtype) {
   continue;
  }

  if ((new_col->prtype & DATA_NOT_NULL)
      && dfield_is_null(dfield)) {

   if (!log->allow_not_null) {
    /* We got a NULL value for a NOT NULL column. */
    *error = DB_INVALID_NULL;
    return NULL;
   }

   const dfield_t& default_field
    = log->defaults->fields[col_no];

   Field* field = log->old_table->field[col->ind];

   field->set_warning(Sql_condition::WARN_LEVEL_WARN,
        WARN_DATA_TRUNCATED, 1,
        ulong(log->n_rows));

   *dfield = default_field;
  }

  /* Adjust the DATA_NOT_NULL flag in the parsed row. */
  dfield_get_type(dfield)->prtype = new_col->prtype;

  ut_ad(dict_col_type_assert_equal(new_col,
       dfield_get_type(dfield)));
 }

 return(row);
}

/******************************************************//**
Replays an insert operation on a table that was rebuilt.
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
row_log_table_apply_insert_low(
/*===========================*/
 que_thr_t*  thr,  /*!< in: query graph */
 const dtuple_t*  row,  /*!< in: table row
in the old table definition */
 mem_heap_t*  offsets_heap, /*!< in/out: memory heap
that can be emptied */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 row_merge_dup_t* dup)  /*!< in/out: for reporting
duplicate key errors */
{
 dberr_t  error;
 dtuple_t* entry;
 const row_log_t*log = dup->index->online_log;
 dict_index_t* index = dict_table_get_first_index(log->table);
 ulint  n_index = 0;

 ut_ad(dtuple_validate(row));

 DBUG_LOG("ib_alter_table",
   "insert table " << index->table->id << " (index "
   << index->id << "): " << rec_printer(row).str());

 static const ulint flags
  = (BTR_CREATE_FLAG
     | BTR_NO_LOCKING_FLAG
     | BTR_NO_UNDO_LOG_FLAG
     | BTR_KEEP_SYS_FLAG);

 entry = row_build_index_entry(row, NULL, index, heap);

 error = row_ins_clust_index_entry_low(
  flags, BTR_MODIFY_TREE, index, index->n_uniq,
  entry, 0, thr);

 switch (error) {
 case DB_SUCCESS:
  break;
 case DB_SUCCESS_LOCKED_REC:
  /* The row had already been copied to the table. */
  return(DB_SUCCESS);
 default:
  return(error);
 }

 ut_ad(dict_index_is_clust(index));

 for (n_index += index->type != DICT_CLUSTERED;
      (index = dict_table_get_next_index(index)); n_index++) {
  if (index->type & DICT_FTS) {
   continue;
  }

  entry = row_build_index_entry(row, NULL, index, heap);
  error = row_ins_sec_index_entry_low(
   flags, BTR_INSERT_TREE,
   index, offsets_heap, heap, entry,
   thr_get_trx(thr)->id, thr);

  if (error != DB_SUCCESS) {
   if (error == DB_DUPLICATE_KEY) {
    thr_get_trx(thr)->error_key_num = n_index;
   }
   break;
  }
 }

 return(error);
}

/******************************************************//**
Replays an insert operation on a table that was rebuilt.
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
row_log_table_apply_insert(
/*=======================*/
 que_thr_t*  thr,  /*!< in: query graph */
 const mrec_t*  mrec,  /*!< in: record to insert */
 const rec_offs*  offsets, /*!< in: offsets of mrec */
 mem_heap_t*  offsets_heap, /*!< in/out: memory heap
that can be emptied */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 row_merge_dup_t* dup)  /*!< in/out: for reporting
duplicate key errors */
{
 row_log_t*log = dup->index->online_log;
 dberr_t  error;
 const dtuple_t* row = row_log_table_apply_convert_mrec(
  mrec, dup->index, offsets, log, heap, &error);

 switch (error) {
 case DB_SUCCESS:
  ut_ad(row != NULL);
  break;
 default:
  ut_ad(0);
  /* fall through */
 case DB_INVALID_NULL:
  ut_ad(row == NULL);
  return(error);
 }

 error = row_log_table_apply_insert_low(
  thr, row, offsets_heap, heap, dup);
 if (error != DB_SUCCESS) {
  /* Report the erroneous row using the new
version of the table. */
  innobase_row_to_mysql(dup->table, log->table, row);
 }
 return(error);
}

/******************************************************//**
Deletes a record from a table that is being rebuilt.
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
row_log_table_apply_delete_low(
/*===========================*/
 btr_pcur_t*  pcur,  /*!< in/out: B-tree cursor,
will be trashed */
 const rec_offs*  offsets, /*!< in: offsets on pcur */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 mtr_t*   mtr)  /*!< in/out: mini-transaction,
will be committed */
{
 dberr_t  error;
 row_ext_t* ext;
 dtuple_t* row;
 dict_index_t* index = pcur->index();

 ut_ad(dict_index_is_clust(index));

 DBUG_LOG("ib_alter_table",
   "delete table " << index->table->id << " (index "
   << index->id << "): "
   << rec_printer(btr_pcur_get_rec(pcur), offsets).str());

 if (dict_table_get_next_index(index)) {
  /* Build a row template for purging secondary index entries. */
  row = row_build(
   ROW_COPY_DATA, index, btr_pcur_get_rec(pcur),
   offsets, NULL, NULL, NULL, &ext, heap);
 } else {
  row = NULL;
 }

 btr_cur_pessimistic_delete(&error, FALSE, btr_pcur_get_btr_cur(pcur),
       BTR_CREATE_FLAG, false, mtr);
 if (error != DB_SUCCESS) {
err_exit:
  mtr->commit();
  return error;
 }

 mtr->commit();

 while ((index = dict_table_get_next_index(index)) != NULL) {
  if (index->type & DICT_FTS) {
   continue;
  }

  const dtuple_t* entry = row_build_index_entry(
   row, ext, index, heap);
  mtr->start();
  index->set_modified(*mtr);
  pcur->btr_cur.page_cur.index = index;
  error = btr_pcur_open(entry, PAGE_CUR_LE, BTR_PURGE_TREE, pcur,
          mtr);
  if (error) {
   goto err_exit;
  }
  ut_ad(pcur->btr_cur.flag == BTR_CUR_BINARY);

  if (page_rec_is_infimum(btr_pcur_get_rec(pcur))
      || btr_pcur_get_low_match(pcur) < index->n_uniq) {
   /* All secondary index entries should be
   found, because new_table is being modified by
   this thread only, and all indexes should be
updated in sync. */
   error = DB_INDEX_CORRUPT;
   goto err_exit;
  }

  btr_cur_pessimistic_delete(&error, FALSE,
        btr_pcur_get_btr_cur(pcur),
        BTR_CREATE_FLAG, false, mtr);
  mtr->commit();
 }

 return(error);
}

/******************************************************//**
Replays a delete operation on a table that was rebuilt.
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
row_log_table_apply_delete(
/*=======================*/
 ulint   trx_id_col, /*!< in: position of
      DB_TRX_ID in the new
clustered index */
 const mrec_t*  mrec,  /*!< in: merge record */
 const rec_offs*  moffsets, /*!< in: offsets of mrec */
 mem_heap_t*  offsets_heap, /*!< in/out: memory heap
that can be emptied */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 const row_merge_dup_t* dup)  /*!< in: context */
{
 const row_log_t* const log{dup->index->online_log};
 dict_table_t* new_table = log->table;
 dict_index_t* index = dict_table_get_first_index(new_table);
 dtuple_t* old_pk;
 mtr_t  mtr{dup->trx};
 btr_pcur_t pcur;
 rec_offs* offsets;

 pcur.btr_cur.page_cur.index = index;
 ut_ad(rec_offs_n_fields(moffsets) == index->first_user_field());
 ut_ad(!rec_offs_any_extern(moffsets));

 /* Convert the row to a search tuple. */
 old_pk = dtuple_create(heap, index->n_uniq);
 dict_index_copy_types(old_pk, index, index->n_uniq);

 for (ulint i = 0; i < index->n_uniq; i++) {
  ulint  len;
  const void* field;
  field = rec_get_nth_field(mrec, moffsets, i, &len);
  ut_ad(len != UNIV_SQL_NULL);
  dfield_set_data(dtuple_get_nth_field(old_pk, i),
    field, len);
 }

 mtr_start(&mtr);
 index->set_modified(mtr);
 dberr_t err = btr_pcur_open(old_pk, PAGE_CUR_LE, BTR_PURGE_TREE, &pcur,
        &mtr);
 if (err != DB_SUCCESS) {
  goto all_done;
 }

 ut_ad(btr_pcur_get_btr_cur(&pcur)->flag == BTR_CUR_BINARY);

 if (page_rec_is_infimum(btr_pcur_get_rec(&pcur))
     || btr_pcur_get_low_match(&pcur) < index->n_uniq) {
all_done:
  mtr_commit(&mtr);
  /* The record was not found. All done. */
  /* This should only happen when an earlier
  ROW_T_INSERT was skipped or
  ROW_T_UPDATE was interpreted as ROW_T_DELETE
due to BLOBs having been freed by rollback. */
  return err;
 }

 offsets = rec_get_offsets(btr_pcur_get_rec(&pcur), index, nullptr,
      index->n_core_fields,
      ULINT_UNDEFINED, &offsets_heap);
#if defined UNIV_DEBUG || defined UNIV_BLOB_LIGHT_DEBUG
 ut_a(!rec_offs_any_null_extern(btr_pcur_get_rec(&pcur), offsets));
#endif /* UNIV_DEBUG || UNIV_BLOB_LIGHT_DEBUG */

 /* Only remove the record if DB_TRX_ID,DB_ROLL_PTR match. */

 {
  ulint  len;
  const byte* mrec_trx_id
   = rec_get_nth_field(mrec, moffsets, trx_id_col, &len);
  ut_ad(len == DATA_TRX_ID_LEN);
  const byte* rec_trx_id
   = rec_get_nth_field(btr_pcur_get_rec(&pcur), offsets,
         trx_id_col, &len);
  ut_ad(len == DATA_TRX_ID_LEN);
  ut_d(trx_id_check(rec_trx_id, log->min_trx));
  ut_d(trx_id_check(mrec_trx_id, log->min_trx));

  ut_ad(rec_get_nth_field(mrec, moffsets, trx_id_col + 1, &len)
        == mrec_trx_id + DATA_TRX_ID_LEN);
  ut_ad(len == DATA_ROLL_PTR_LEN);
  ut_ad(rec_get_nth_field(btr_pcur_get_rec(&pcur), offsets,
     trx_id_col + 1, &len)
        == rec_trx_id + DATA_TRX_ID_LEN);
  ut_ad(len == DATA_ROLL_PTR_LEN);

  if (memcmp(mrec_trx_id, rec_trx_id,
      DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN)) {
   /* The ROW_T_DELETE was logged for a different
   PRIMARY KEY,DB_TRX_ID,DB_ROLL_PTR.
   This is possible if a ROW_T_INSERT was skipped
   or a ROW_T_UPDATE was interpreted as ROW_T_DELETE
   because some BLOBs were missing due to
   (1) rolling back the initial insert, or
   (2) purging the BLOB for a later ROW_T_DELETE
   (3) purging 'old values' for a later ROW_T_UPDATE
or ROW_T_DELETE. */
   ut_ad(!log->same_pk);
   goto all_done;
  }
 }

 return row_log_table_apply_delete_low(&pcur, offsets, heap, &mtr);
}

/******************************************************//**
Replays an update operation on a table that was rebuilt.
@return DB_SUCCESS or error code */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
dberr_t
row_log_table_apply_update(
/*=======================*/
 que_thr_t*  thr,  /*!< in: query graph */
 ulint   new_trx_id_col, /*!< in: position of
      DB_TRX_ID in the new
clustered index */
 const mrec_t*  mrec,  /*!< in: new value */
 const rec_offs*  offsets, /*!< in: offsets of mrec */
 mem_heap_t*  offsets_heap, /*!< in/out: memory heap
that can be emptied */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 row_merge_dup_t* dup,  /*!< in/out: for reporting
duplicate key errors */
 const dtuple_t*  old_pk)  /*!< in: PRIMARY KEY and
      DB_TRX_ID,DB_ROLL_PTR
      of the old value,
or PRIMARY KEY if same_pk */
{
 row_log_t* log = dup->index->online_log;
 const dtuple_t* row;
 dict_index_t* index = dict_table_get_first_index(log->table);
 mtr_t  mtr{dup->trx};
 btr_pcur_t pcur;
 dberr_t  error;
 ulint  n_index = 0;

 pcur.btr_cur.page_cur.index = index;

 ut_ad(dtuple_get_n_fields_cmp(old_pk)
       == dict_index_get_n_unique(index));
 ut_ad(dtuple_get_n_fields(old_pk) - (log->same_pk ? 0 : 2)
       == dict_index_get_n_unique(index));

 row = row_log_table_apply_convert_mrec(
  mrec, dup->index, offsets, log, heap, &error);

 switch (error) {
 case DB_SUCCESS:
  ut_ad(row != NULL);
  break;
 default:
  ut_ad(0);
  /* fall through */
 case DB_INVALID_NULL:
  ut_ad(row == NULL);
  return(error);
 }

 mtr.start();
 index->set_modified(mtr);
 error = btr_pcur_open(old_pk, PAGE_CUR_LE, BTR_MODIFY_TREE, &pcur,
         &mtr);
 if (error != DB_SUCCESS) {
func_exit:
  mtr.commit();
func_exit_committed:
  ut_ad(mtr.has_committed());
  ut_free(pcur.old_rec_buf);

  if (error != DB_SUCCESS) {
   /* Report the erroneous row using the new
version of the table. */
   innobase_row_to_mysql(dup->table, log->table, row);
  }

  return error;
 }

 ut_ad(btr_pcur_get_btr_cur(&pcur)->flag == BTR_CUR_BINARY);

 ut_ad(!page_rec_is_infimum(btr_pcur_get_rec(&pcur))
       && btr_pcur_get_low_match(&pcur) >= index->n_uniq);

 /* Prepare to update (or delete) the record. */
 rec_offs*  cur_offsets = rec_get_offsets(
  btr_pcur_get_rec(&pcur), index, nullptr, index->n_core_fields,
  ULINT_UNDEFINED, &offsets_heap);

#ifdef UNIV_DEBUG
 if (!log->same_pk) {
  ulint  len;
  const byte* rec_trx_id
   = rec_get_nth_field(btr_pcur_get_rec(&pcur),
         cur_offsets, index->n_uniq, &len);
  const dfield_t* old_pk_trx_id
   = dtuple_get_nth_field(old_pk, index->n_uniq);
  ut_ad(len == DATA_TRX_ID_LEN);
  ut_d(trx_id_check(rec_trx_id, log->min_trx));
  ut_ad(old_pk_trx_id->len == DATA_TRX_ID_LEN);
  ut_ad(old_pk_trx_id[1].len == DATA_ROLL_PTR_LEN);
  ut_ad(DATA_TRX_ID_LEN
        + static_cast<const char*>(old_pk_trx_id->data)
        == old_pk_trx_id[1].data);
  ut_d(trx_id_check(old_pk_trx_id->data, log->min_trx));
  ut_ad(!memcmp(rec_trx_id, old_pk_trx_id->data,
         DATA_TRX_ID_LEN + DATA_ROLL_PTR_LEN));
 }
#endif

 dtuple_t* entry = row_build_index_entry_low(
  row, NULL, index, heap, ROW_BUILD_NORMAL);
 upd_t*  update = row_upd_build_difference_binary(
  index, entry, btr_pcur_get_rec(&pcur), cur_offsets,
  false, false, NULL, heap, dup->table, &error);
 if (error != DB_SUCCESS || !update->n_fields) {
  goto func_exit;
 }

 const bool pk_updated
  = upd_get_nth_field(update, 0)->field_no < new_trx_id_col;

 if (pk_updated || rec_offs_any_extern(cur_offsets)) {
  /* If the record contains any externally stored
  columns, perform the update by delete and insert,
  because we will not write any undo log that would
  allow purge to free any orphaned externally stored
columns. */

  if (pk_updated && log->same_pk) {
   /* The ROW_T_UPDATE log record should only be
   written when the PRIMARY KEY fields of the
   record did not change in the old table.  We
   can only get a change of PRIMARY KEY columns
   in the rebuilt table if the PRIMARY KEY was
redefined (!same_pk). */
   ut_ad(0);
   error = DB_CORRUPTION;
   goto func_exit;
  }

  error = row_log_table_apply_delete_low(
   &pcur, cur_offsets, heap, &mtr);
  ut_ad(mtr.has_committed());

  if (error == DB_SUCCESS) {
   error = row_log_table_apply_insert_low(
    thr, row, offsets_heap, heap, dup);
  }

  goto func_exit_committed;
 }

 dtuple_t* old_row;
 row_ext_t* old_ext;

 if (dict_table_get_next_index(index)) {
  /* Construct the row corresponding to the old value of
the record. */
  old_row = row_build(
   ROW_COPY_DATA, index, btr_pcur_get_rec(&pcur),
   cur_offsets, NULL, NULL, NULL, &old_ext, heap);
  ut_ad(old_row);

  DBUG_LOG("ib_alter_table",
    "update table " << index->table->id
    << " (index " << index->id
    << ": " << rec_printer(old_row).str()
    << " to " << rec_printer(row).str());
 } else {
  old_row = NULL;
  old_ext = NULL;
 }

 big_rec_t* big_rec;

 error = btr_cur_pessimistic_update(
  BTR_CREATE_FLAG | BTR_NO_LOCKING_FLAG
  | BTR_NO_UNDO_LOG_FLAG | BTR_KEEP_SYS_FLAG
  | BTR_KEEP_POS_FLAG,
  btr_pcur_get_btr_cur(&pcur),
  &cur_offsets, &offsets_heap, heap, &big_rec,
  update, 0, thr, 0, &mtr);

 if (big_rec) {
  if (error == DB_SUCCESS) {
   error = btr_store_big_rec_extern_fields(
    &pcur, cur_offsets, big_rec, &mtr,
    BTR_STORE_UPDATE);
  }

  dtuple_big_rec_free(big_rec);
 }

 for (n_index += index->type != DICT_CLUSTERED;
      (index = dict_table_get_next_index(index)); n_index++) {
  if (!index->is_btree()) {
   continue;
  }

  if (error != DB_SUCCESS) {
   break;
  }

  if (!row_upd_changes_ord_field_binary(
       index, update, thr, old_row, NULL)) {
   continue;
  }

  if (dict_index_has_virtual(index)) {
   dtuple_copy_v_fields(old_row, old_pk);
  }

  mtr.commit();

  entry = row_build_index_entry(old_row, old_ext, index, heap);
  if (!entry) {
   ut_ad(0);
   error = DB_CORRUPTION;
   goto func_exit_committed;
  }

  mtr.start();
  index->set_modified(mtr);
  pcur.btr_cur.page_cur.index = index;

  ut_free(pcur.old_rec_buf);
  pcur.old_rec_buf = nullptr;

  error = btr_pcur_open(entry, PAGE_CUR_LE, BTR_MODIFY_TREE,
          &pcur, &mtr);

  if (error != DB_SUCCESS) {
   ut_ad(0);
   break;
  }

  if (btr_pcur_is_before_first_on_page(&pcur)
      || btr_pcur_get_low_match(&pcur)
      != dtuple_get_n_fields(entry)) {
   ut_ad(0);
   error = DB_CORRUPTION;
   break;
  }

  btr_cur_pessimistic_delete(
   &error, FALSE, btr_pcur_get_btr_cur(&pcur),
   BTR_CREATE_FLAG, false, &mtr);

  if (error != DB_SUCCESS) {
   break;
  }

  mtr.commit();

  entry = row_build_index_entry(row, NULL, index, heap);
  error = row_ins_sec_index_entry_low(
   BTR_CREATE_FLAG | BTR_NO_LOCKING_FLAG
   | BTR_NO_UNDO_LOG_FLAG | BTR_KEEP_SYS_FLAG,
   BTR_INSERT_TREE, index, offsets_heap, heap,
   entry, thr_get_trx(thr)->id, thr);

  /* Report correct index name for duplicate key error. */
  if (error == DB_DUPLICATE_KEY) {
   thr_get_trx(thr)->error_key_num = n_index;
  }

  mtr.start();
  index->set_modified(mtr);
 }

 goto func_exit;
}

/******************************************************//**
Applies an operation to a table that was rebuilt.
@return NULL on failure (mrec corruption) or when out of data;
pointer to next record on success */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
const mrec_t*
row_log_table_apply_op(
/*===================*/
 que_thr_t*  thr,  /*!< in: query graph */
 ulint   new_trx_id_col, /*!< in: position of
DB_TRX_ID in new index */
 row_merge_dup_t* dup,  /*!< in/out: for reporting
duplicate key errors */
 dberr_t*  error,  /*!< out: DB_SUCCESS
or error code */
 mem_heap_t*  offsets_heap, /*!< in/out: memory heap
that can be emptied */
 mem_heap_t*  heap,  /*!< in/out: memory heap */
 const mrec_t*  mrec,  /*!< in: merge record */
 const mrec_t*  mrec_end, /*!< in: end of buffer */
 rec_offs*  offsets) /*!< in/out: work area
for parsing mrec */
{
 row_log_t* log = dup->index->online_log;
 dict_index_t* new_index = dict_table_get_first_index(log->table);
 ulint  extra_size;
 const mrec_t* next_mrec;
 dtuple_t* old_pk;

 ut_ad(dict_index_is_clust(dup->index));
 ut_ad(dup->index->table != log->table);
 ut_ad(log->head.total <= log->tail.total);

 *error = DB_SUCCESS;

 /* 3 = 1 (op type) + 1 (extra_size) + at least 1 byte payload */
 if (mrec + 3 >= mrec_end) {
  return(NULL);
 }

 const bool is_instant = log->is_instant(dup->index);
 const mrec_t* const mrec_start = mrec;

 switch (*mrec++) {
 default:
  ut_ad(0);
  *error = DB_CORRUPTION;
  return(NULL);
 case ROW_T_INSERT:
  extra_size = *mrec++;

  if (extra_size >= 0x80) {
   /* Read another byte of extra_size. */

   extra_size = (extra_size & 0x7f) << 8;
   extra_size |= *mrec++;
  }

  mrec += extra_size;

  ut_ad(extra_size || !is_instant);

  if (mrec > mrec_end) {
   return(NULL);
  }

  rec_offs_set_n_fields(offsets, dup->index->n_fields);
  rec_init_offsets_temp(mrec, dup->index, offsets,
          log->n_core_fields, log->non_core_fields,
          is_instant
          ? static_cast<rec_comp_status_t>(
           *(mrec - extra_size))
          : REC_STATUS_ORDINARY);

  next_mrec = mrec + rec_offs_data_size(offsets);

  if (next_mrec > mrec_end) {
   return(NULL);
  } else {
   log->head.total += ulint(next_mrec - mrec_start);
   *error = row_log_table_apply_insert(
    thr, mrec, offsets, offsets_heap,
    heap, dup);
  }
  break;

 case ROW_T_DELETE:
  extra_size = *mrec++;
  ut_ad(mrec < mrec_end);

  /* We assume extra_size < 0x100 for the PRIMARY KEY prefix.
For fixed-length PRIMARY key columns, it is 0. */
  mrec += extra_size;

  /* The ROW_T_DELETE record was converted by
rec_convert_dtuple_to_temp() using new_index. */
  ut_ad(!new_index->is_instant());
  rec_offs_set_n_fields(offsets, new_index->first_user_field());
  rec_init_offsets_temp(mrec, new_index, offsets);
  next_mrec = mrec + rec_offs_data_size(offsets);
  if (next_mrec > mrec_end) {
   return(NULL);
  }

  log->head.total += ulint(next_mrec - mrec_start);

  *error = row_log_table_apply_delete(
   new_trx_id_col,
   mrec, offsets, offsets_heap, heap, dup);
  break;

 case ROW_T_UPDATE:
  /* Logically, the log entry consists of the
  (PRIMARY KEY,DB_TRX_ID) of the old value (converted
  to the new primary key definition) followed by
  the new value in the old table definition. If the
  definition of the columns belonging to PRIMARY KEY
  is not changed, the log will only contain
DB_TRX_ID,new_row. */

  if (log->same_pk) {
   ut_ad(new_index->n_uniq == dup->index->n_uniq);

   extra_size = *mrec++;

   if (extra_size >= 0x80) {
    /* Read another byte of extra_size. */

    extra_size = (extra_size & 0x7f) << 8;
    extra_size |= *mrec++;
   }

   mrec += extra_size;

   ut_ad(extra_size || !is_instant);

   if (mrec > mrec_end) {
    return(NULL);
   }

   rec_offs_set_n_fields(offsets, dup->index->n_fields);
   rec_init_offsets_temp(mrec, dup->index, offsets,
           log->n_core_fields,
           log->non_core_fields,
           is_instant
           ? static_cast<rec_comp_status_t>(
            *(mrec - extra_size))
           : REC_STATUS_ORDINARY);

   next_mrec = mrec + rec_offs_data_size(offsets);

   if (next_mrec > mrec_end) {
    return(NULL);
   }

   old_pk = dtuple_create(heap, new_index->n_uniq);
   dict_index_copy_types(
    old_pk, new_index, old_pk->n_fields);

   /* Copy the PRIMARY KEY fields from mrec to old_pk. */
   for (ulint i = 0; i < new_index->n_uniq; i++) {
    const void* field;
    ulint  len;
    dfield_t* dfield;

    ut_ad(!rec_offs_nth_extern(offsets, i));

    field = rec_get_nth_field(
     mrec, offsets, i, &len);
    ut_ad(len != UNIV_SQL_NULL);

    dfield = dtuple_get_nth_field(old_pk, i);
    dfield_set_data(dfield, field, len);
   }
  } else {
   /* We assume extra_size < 0x100
for the PRIMARY KEY prefix. */
   mrec += *mrec + 1;

   if (mrec > mrec_end) {
    return(NULL);
   }

   /* Get offsets for PRIMARY KEY,
DB_TRX_ID, DB_ROLL_PTR. */
   /* The old_pk prefix was converted by
rec_convert_dtuple_to_temp() using new_index. */
   ut_ad(!new_index->is_instant());
   rec_offs_set_n_fields(offsets,
           new_index->first_user_field());
   rec_init_offsets_temp(mrec, new_index, offsets);

   next_mrec = mrec + rec_offs_data_size(offsets);
   if (next_mrec + 2 > mrec_end) {
    return(NULL);
   }

   /* Copy the PRIMARY KEY fields and
DB_TRX_ID, DB_ROLL_PTR from mrec to old_pk. */
   old_pk = dtuple_create(heap,
            new_index->first_user_field());
   dict_index_copy_types(old_pk, new_index,
           old_pk->n_fields);

   for (ulint i = 0; i < new_index->first_user_field();
        i++) {
    const void* field;
    ulint  len;
    dfield_t* dfield;

    ut_ad(!rec_offs_nth_extern(offsets, i));

    field = rec_get_nth_field(
     mrec, offsets, i, &len);
    ut_ad(len != UNIV_SQL_NULL);

    dfield = dtuple_get_nth_field(old_pk, i);
    dfield_set_data(dfield, field, len);
   }

   mrec = next_mrec;

   /* Fetch the new value of the row as it was
in the old table definition. */
   extra_size = *mrec++;

   if (extra_size >= 0x80) {
    /* Read another byte of extra_size. */

    extra_size = (extra_size & 0x7f) << 8;
    extra_size |= *mrec++;
   }

   mrec += extra_size;

   ut_ad(extra_size || !is_instant);

   if (mrec > mrec_end) {
    return(NULL);
   }

   rec_offs_set_n_fields(offsets, dup->index->n_fields);
   rec_init_offsets_temp(mrec, dup->index, offsets,
           log->n_core_fields,
           log->non_core_fields,
           is_instant
           ? static_cast<rec_comp_status_t>(
            *(mrec - extra_size))
           : REC_STATUS_ORDINARY);

   next_mrec = mrec + rec_offs_data_size(offsets);

   if (next_mrec > mrec_end) {
    return(NULL);
   }
  }

  ut_ad(next_mrec <= mrec_end);
  log->head.total += ulint(next_mrec - mrec_start);
  dtuple_set_n_fields_cmp(old_pk, new_index->n_uniq);

  *error = row_log_table_apply_update(
   thr, new_trx_id_col,
   mrec, offsets, offsets_heap, heap, dup, old_pk);
  break;
 }

 ut_ad(log->head.total <= log->tail.total);
 mem_heap_empty(offsets_heap);
 mem_heap_empty(heap);
 return(next_mrec);
}

#ifdef HAVE_PSI_STAGE_INTERFACE
/** Estimate how much an ALTER TABLE progress should be incremented per
one block of log applied.
For the other phases of ALTER TABLE we increment the progress with 1 per
page processed.
@return amount of abstract units to add to work_completed when one block
of log is applied.
*/
inline
ulint
row_log_progress_inc_per_block()
{
 /* We must increment the progress once per page (as in
 srv_page_size, default = innodb_page_size=16KiB).
One block here is srv_sort_buf_size (usually 1MiB). */
 const ulint pages_per_block = std::max<ulint>(
  ulint(srv_sort_buf_size >> srv_page_size_shift), 1);

 /* Multiply by an artificial factor of 6 to even the pace with
 the rest of the ALTER TABLE phases, they process page_size amount
of data faster. */
 return(pages_per_block * 6);
}

/** Estimate how much work is to be done by the log apply phase
of an ALTER TABLE for this index.
@param[in] index index whose log to assess
@return work to be done by log-apply in abstract units
*/
ulint
row_log_estimate_work(
 const dict_index_t* index)
{
 if (index == NULL || index->online_log == NULL
     || index->online_log_is_dummy()) {
  return(0);
 }

 const row_log_t* l = index->online_log;
 const ulint  bytes_left =
  static_cast<ulint>(l->tail.total - l->head.total);
 const ulint  blocks_left = bytes_left / srv_sort_buf_size;

 return(blocks_left * row_log_progress_inc_per_block());
}
#else /* HAVE_PSI_STAGE_INTERFACE */
inline
ulint
row_log_progress_inc_per_block()
{
 return(0);
}
#endif /* HAVE_PSI_STAGE_INTERFACE */

/** Applies operations to a table was rebuilt.
@param[in] thr query graph
@param[in,out] dup for reporting duplicate key errors
@param[in,out] stage performance schema accounting object, used by
ALTER TABLE. If not NULL, then stage->inc() will be called for each block
of log that is applied.
@return DB_SUCCESS, or error code on failure */
static MY_ATTRIBUTE((warn_unused_result))
dberr_t
row_log_table_apply_ops(
 que_thr_t*  thr,
 row_merge_dup_t* dup,
 ut_stage_alter_t* stage)
{
 dberr_t  error;
 const mrec_t* mrec  = NULL;
 const mrec_t* next_mrec;
 const mrec_t* mrec_end = NULL; /* silence bogus warning */
 const mrec_t* next_mrec_end;
 mem_heap_t* heap;
 mem_heap_t* offsets_heap;
 rec_offs* offsets;
 bool  has_index_lock;
 dict_index_t* index  = const_cast<dict_index_t*>(
  dup->index);
 dict_table_t* new_table = index->online_log->table;
 dict_index_t* new_index = dict_table_get_first_index(
  new_table);
 const ulint i  = 1 + REC_OFFS_HEADER_SIZE
  + std::max<ulint>(index->n_fields,
      new_index->first_user_field());
 const ulint new_trx_id_col = dict_col_get_clust_pos(
  dict_table_get_sys_col(new_table, DATA_TRX_ID), new_index);
 trx_t*  trx  = thr_get_trx(thr);

 ut_ad(dict_index_is_clust(index));
 ut_ad(dict_index_is_online_ddl(index));
 ut_ad(trx->mysql_thd);
 ut_ad(index->lock.have_x());
 ut_ad(!dict_index_is_online_ddl(new_index));
 ut_ad(dict_col_get_clust_pos(
        dict_table_get_sys_col(index->table, DATA_TRX_ID), index)
       != ULINT_UNDEFINED);
 ut_ad(new_trx_id_col > 0);
 ut_ad(new_trx_id_col != ULINT_UNDEFINED);

 MEM_UNDEFINED(&mrec_end, sizeof mrec_end);

 offsets = static_cast<rec_offs*>(ut_malloc_nokey(i * sizeof *offsets));
 rec_offs_set_n_alloc(offsets, i);
 rec_offs_set_n_fields(offsets, dict_index_get_n_fields(index));

 heap = mem_heap_create(srv_page_size);
 offsets_heap = mem_heap_create(srv_page_size);
 has_index_lock = true;

next_block:
 ut_ad(has_index_lock);
 ut_ad(index->lock.have_u_or_x());
 ut_ad(index->online_log->head.bytes == 0);

 stage->inc(row_log_progress_inc_per_block());

 if (trx_is_interrupted(trx)) {
  goto interrupted;
 }

 if (index->is_corrupted()) {
  error = DB_INDEX_CORRUPT;
  goto func_exit;
 }

 ut_ad(dict_index_is_online_ddl(index));

 error = index->online_log->error;

 if (error != DB_SUCCESS) {
  goto func_exit;
 }

 if (UNIV_UNLIKELY(index->online_log->head.blocks
     > index->online_log->tail.blocks)) {
unexpected_eof:
  ib::error() << "Unexpected end of temporary file for table "
   << index->table->name;
corruption:
  error = DB_CORRUPTION;
  goto func_exit;
 }

 if (index->online_log->head.blocks
     == index->online_log->tail.blocks) {
  if (index->online_log->head.blocks) {
#ifdef HAVE_FTRUNCATE
   /* Truncate the file in order to save space. */
   if (index->online_log->fd > 0
       && ftruncate(index->online_log->fd, 0) == -1) {
    ib::error()
     << "\'" << index->name + 1
     << "\' failed with error "
     << errno << ":" << strerror(errno);

    goto corruption;
   }
#endif /* HAVE_FTRUNCATE */
   index->online_log->head.blocks
    = index->online_log->tail.blocks = 0;
  }

  next_mrec = index->online_log->tail.block;
  next_mrec_end = next_mrec + index->online_log->tail.bytes;

  if (next_mrec_end == next_mrec) {
   /* End of log reached. */
all_done:
   ut_ad(has_index_lock);
   ut_ad(index->online_log->head.blocks == 0);
   ut_ad(index->online_log->tail.blocks == 0);
   index->online_log->head.bytes = 0;
   index->online_log->tail.bytes = 0;
   error = DB_SUCCESS;
   goto func_exit;
  }
 } else {
  os_offset_t ofs;

  ofs = (os_offset_t) index->online_log->head.blocks
   * srv_sort_buf_size;

  ut_ad(has_index_lock);
  has_index_lock = false;
  index->lock.x_unlock();

  log_free_check();

  ut_ad(dict_index_is_online_ddl(index));

  if (!row_log_block_allocate(index->online_log->head)) {
   error = DB_OUT_OF_MEMORY;
   goto func_exit;
  }

  byte*   buf = index->online_log->head.block;

  if (DB_SUCCESS
      != os_file_read(IORequestRead, index->online_log->fd,
        buf, ofs, srv_sort_buf_size, nullptr)) {
   ib::error()
    << "Unable to read temporary file"
    " for table " << index->table->name;
   goto corruption;
  }

  if (srv_encrypt_log) {
   if (!log_tmp_block_decrypt(
        buf, srv_sort_buf_size,
        index->online_log->crypt_head, ofs)) {
    error = DB_DECRYPTION_FAILED;
    goto func_exit;
   }

   srv_stats.n_rowlog_blocks_decrypted.inc();
   memcpy(buf, index->online_log->crypt_head,
          srv_sort_buf_size);
  }

#ifdef POSIX_FADV_DONTNEED
  /* Each block is read exactly once.  Free up the file cache. */
  posix_fadvise(index->online_log->fd,
         ofs, srv_sort_buf_size, POSIX_FADV_DONTNEED);
#endif /* POSIX_FADV_DONTNEED */

  next_mrec = index->online_log->head.block;
  next_mrec_end = next_mrec + srv_sort_buf_size;
 }

 /* This read is not protected by index->online_log->mutex for
 performance reasons. We will eventually notice any error that
was flagged by a DML thread. */
 error = index->online_log->error;

 if (error != DB_SUCCESS) {
  goto func_exit;
 }

 if (mrec) {
  /* A partial record was read from the previous block.
  Copy the temporary buffer full, as we do not know the
  length of the record. Parse subsequent records from
  the bigger buffer index->online_log->head.block
or index->online_log->tail.block. */

  ut_ad(mrec == index->online_log->head.buf);
  ut_ad(mrec_end > mrec);
  ut_ad(mrec_end < (&index->online_log->head.buf)[1]);

  memcpy((mrec_t*) mrec_end, next_mrec,
         ulint((&index->online_log->head.buf)[1] - mrec_end));
  mrec = row_log_table_apply_op(
   thr, new_trx_id_col,
   dup, &error, offsets_heap, heap,
   index->online_log->head.buf,
   (&index->online_log->head.buf)[1], offsets);
  if (error != DB_SUCCESS) {
   goto func_exit;
  } else if (UNIV_UNLIKELY(mrec == NULL)) {
   /* The record was not reassembled properly. */
   goto corruption;
  }
  /* The record was previously found out to be
  truncated. Now that the parse buffer was extended,
it should proceed beyond the old end of the buffer. */
  ut_a(mrec > mrec_end);

  index->online_log->head.bytes = ulint(mrec - mrec_end);
  next_mrec += index->online_log->head.bytes;
 }

 ut_ad(next_mrec <= next_mrec_end);
 /* The following loop must not be parsing the temporary
buffer, but head.block or tail.block. */

 /* mrec!=NULL means that the next record starts from the
middle of the block */
 ut_ad((mrec == NULL) == (index->online_log->head.bytes == 0));

#ifdef UNIV_DEBUG
 if (next_mrec_end - srv_sort_buf_size
     == index->online_log->head.block) {
  /* If tail.bytes == 0, next_mrec_end can also be at
the end of tail.block. */
  if (index->online_log->tail.bytes == 0) {
   ut_ad(next_mrec == next_mrec_end);
   ut_ad(index->online_log->tail.blocks == 0);
   ut_ad(index->online_log->head.blocks == 0);
   ut_ad(index->online_log->head.bytes == 0);
  } else {
   ut_ad(next_mrec == index->online_log->head.block
         + index->online_log->head.bytes);
   ut_ad(index->online_log->tail.blocks
         > index->online_log->head.blocks);
  }
 } else if (next_mrec_end - index->online_log->tail.bytes
     == index->online_log->tail.block) {
  ut_ad(next_mrec == index->online_log->tail.block
        + index->online_log->head.bytes);
  ut_ad(index->online_log->tail.blocks == 0);
  ut_ad(index->online_log->head.blocks == 0);
  ut_ad(index->online_log->head.bytes
        <= index->online_log->tail.bytes);
 } else {
  ut_error;
 }
#endif /* UNIV_DEBUG */

 mrec_end = next_mrec_end;

 while (!trx_is_interrupted(trx)) {
  mrec = next_mrec;
  ut_ad(mrec <= mrec_end);

  if (mrec == mrec_end) {
   /* We are at the end of the log.
   Mark the replay all_done. */
   if (has_index_lock) {
    goto all_done;
   }
  }

  if (!has_index_lock) {
   /* We are applying operations from a different
   block than the one that is being written to.
   We do not hold index->lock in order to
   allow other threads to concurrently buffer
modifications. */
   ut_ad(mrec >= index->online_log->head.block);
   ut_ad(mrec_end == index->online_log->head.block
         + srv_sort_buf_size);
   ut_ad(index->online_log->head.bytes
         < srv_sort_buf_size);

   /* Take the opportunity to do a redo log
checkpoint if needed. */
   log_free_check();
  } else {
   /* We are applying operations from the last block.
   Do not allow other threads to buffer anything,
so that we can finally catch up and synchronize. */
   ut_ad(index->online_log->head.blocks == 0);
   ut_ad(index->online_log->tail.blocks == 0);
   ut_ad(mrec_end == index->online_log->tail.block
         + index->online_log->tail.bytes);
   ut_ad(mrec >= index->online_log->tail.block);
  }

  /* This read is not protected by index->online_log->mutex
  for performance reasons. We will eventually notice any
error that was flagged by a DML thread. */
  error = index->online_log->error;

  if (error != DB_SUCCESS) {
   goto func_exit;
  }

  next_mrec = row_log_table_apply_op(
   thr, new_trx_id_col,
   dup, &error, offsets_heap, heap,
   mrec, mrec_end, offsets);

  if (error != DB_SUCCESS) {
   goto func_exit;
  } else if (next_mrec == next_mrec_end) {
   /* The record happened to end on a block boundary.
Do we have more blocks left? */
   if (has_index_lock) {
    /* The index will be locked while
applying the last block. */
    goto all_done;
   }

   mrec = NULL;
process_next_block:
   index->lock.x_lock(SRW_LOCK_CALL);
   has_index_lock = true;

   index->online_log->head.bytes = 0;
   index->online_log->head.blocks++;
   goto next_block;
  } else if (next_mrec != NULL) {
   ut_ad(next_mrec < next_mrec_end);
   index->online_log->head.bytes
    += ulint(next_mrec - mrec);
  } else if (has_index_lock) {
   /* When mrec is within tail.block, it should
   be a complete record, because we are holding
index->lock and thus excluding the writer. */
   ut_ad(index->online_log->tail.blocks == 0);
   ut_ad(mrec_end == index->online_log->tail.block
         + index->online_log->tail.bytes);
   ut_ad(0);
   goto unexpected_eof;
  } else {
   const size_t s = size_t(mrec_end - mrec);
   memcpy(index->online_log->head.buf, mrec, s);
   mrec = index->online_log->head.buf;
   mrec_end = mrec + s;
   goto process_next_block;
  }
 }

interrupted:
 error = DB_INTERRUPTED;
func_exit:
 if (!has_index_lock) {
  index->lock.x_lock(SRW_LOCK_CALL);
 }

 mem_heap_free(offsets_heap);
 mem_heap_free(heap);
 row_log_block_free(index->online_log->head);
 ut_free(offsets);
 return(error);
}

/** Apply the row_log_table log to a table upon completing rebuild.
@param[in] thr  query graph
@param[in] old_table old table
@param[in,out] table  MySQL table (for reporting duplicates)
@param[in,out] stage  performance schema accounting object, used by
ALTER TABLE. stage->begin_phase_log_table() will be called initially and then
stage->inc() will be called for each block of log that is applied.
@param[in] new_table Altered table
@return DB_SUCCESS, or error code on failure */
dberr_t
row_log_table_apply(
 que_thr_t*  thr,
 dict_table_t*  old_table,
 struct TABLE*  table,
 ut_stage_alter_t* stage,
 dict_table_t*  new_table)
{
 dberr_t  error;
 dict_index_t* clust_index;
        trx_t* const trx{thr_get_trx(thr)};

 trx->error_key_num = 0;
 DBUG_EXECUTE_IF("innodb_trx_duplicates",
   trx->duplicates = TRX_DUP_REPLACE;);

 stage->begin_phase_log_table();

 clust_index = dict_table_get_first_index(old_table);

 if (clust_index->online_log->n_rows == 0) {
  clust_index->online_log->n_rows = new_table->stat_n_rows;
 }

 clust_index->lock.x_lock(SRW_LOCK_CALL);

 if (!clust_index->online_log) {
  ut_ad(dict_index_get_online_status(clust_index)
        == ONLINE_INDEX_COMPLETE);
  /* This function should not be called unless
  rebuilding a table online. Build in some fault
tolerance. */
  ut_ad(0);
  error = DB_ERROR;
 } else {
  row_merge_dup_t dup = {
   clust_index, trx, table,
   clust_index->online_log->col_map, 0
  };

  error = row_log_table_apply_ops(thr, &dup, stage);

  ut_ad(error != DB_SUCCESS
        || clust_index->online_log->head.total
        == clust_index->online_log->tail.total);
 }

 clust_index->lock.x_unlock();
 DBUG_EXECUTE_IF("innodb_trx_duplicates", trx->duplicates = 0;);

 return(error);
}

/******************************************************//**
Allocate the row log for an index and flag the index
for online creation.
@retval true if success, false if not */
bool
row_log_allocate(
/*=============*/
 const trx_t* trx, /*!< in: the ALTER TABLE transaction */
 dict_index_t* index, /*!< in/out: index */
 dict_table_t* table, /*!< in/out: new table being rebuilt,
or NULL when creating a secondary index */
 bool  same_pk,/*!< in: whether the definition of the
PRIMARY KEY has remained the same */
 const dtuple_t* defaults,
    /*!< in: default values of
added, changed columns, or NULL */
 const ulint* col_map,/*!< in: mapping of old column
numbers to new ones, or NULL if !table */
 const char* path, /*!< in: where to create temporary file */
 const TABLE* old_table, /*!< in: table definition before alter */
 const bool allow_not_null) /*!< in: allow null to not-null
conversion */
{
 row_log_t* log;
 DBUG_ENTER("row_log_allocate");

 ut_ad(!dict_index_is_online_ddl(index));
 ut_ad(dict_index_is_clust(index) == !!table);
 ut_ad(!table || index->table != table);
 ut_ad(same_pk || table);
 ut_ad(!table || col_map);
 ut_ad(!defaults || col_map);
 ut_ad(index->lock.have_u_or_x());
 ut_ad(trx_state_eq(trx, TRX_STATE_ACTIVE));
 ut_ad(trx->id);

 log = static_cast<row_log_t*>(ut_malloc_nokey(sizeof *log));

 if (log == NULL) {
  DBUG_RETURN(false);
 }

 log->fd = OS_FILE_CLOSED;
 mysql_mutex_init(index_online_log_key, &log->mutex, nullptr);

 log->table = table;
 log->same_pk = same_pk;
 log->defaults = defaults;
 log->col_map = col_map;
 log->error = DB_SUCCESS;
 log->min_trx = trx->id;
 log->max_trx = 0;
 log->tail.blocks = log->tail.bytes = 0;
 log->tail.total = 0;
 log->tail.block = log->head.block = NULL;
 log->crypt_tail = log->crypt_head = NULL;
 log->head.blocks = log->head.bytes = 0;
 log->head.total = 0;
 log->path = path;
 log->n_core_fields = index->n_core_fields;
 ut_ad(!table || log->is_instant(index)
       == (index->n_core_fields < index->n_fields));
 log->allow_not_null = allow_not_null;
 log->old_table = old_table;
 log->n_rows = 0;

 if (table && index->is_instant()) {
  const unsigned n = log->n_core_fields;
  log->non_core_fields = UT_NEW_ARRAY_NOKEY(
   dict_col_t::def_t, index->n_fields - n);
  for (unsigned i = n; i < index->n_fields; i++) {
   log->non_core_fields[i - n]
    = index->fields[i].col->def_val;
  }
 } else {
  log->non_core_fields = NULL;
 }

 dict_index_set_online_status(index, ONLINE_INDEX_CREATION);

 if (srv_encrypt_log) {
  log->crypt_head_size = log->crypt_tail_size = srv_sort_buf_size;
  log->crypt_head = static_cast<byte *>(
   my_large_malloc(&log->crypt_head_size, MYF(MY_WME)));
  log->crypt_tail = static_cast<byte *>(
   my_large_malloc(&log->crypt_tail_size, MYF(MY_WME)));

  if (!log->crypt_head || !log->crypt_tail) {
   row_log_free(log);
   DBUG_RETURN(false);
  }
 }

 index->online_log = log;

 if (!table) {
  /* Assign the clustered index online log to table.
  It can be used by concurrent DML to identify whether
the table has any online DDL */
  index->table->indexes.start->online_log_make_dummy();
  log->alter_trx = trx;
 }

 /* While we might be holding an exclusive data dictionary lock
 here, in row_log_abort_sec() we will not always be holding it. Use
atomic operations in both cases. */
 MONITOR_ATOMIC_INC(MONITOR_ONLINE_CREATE_INDEX);

 DBUG_RETURN(true);
}

/******************************************************//**
Free the row log for an index that was being created online. */
void
row_log_free(
/*=========*/
 row_log_t* log) /*!< in,own: row log */
{
 MONITOR_ATOMIC_DEC(MONITOR_ONLINE_CREATE_INDEX);

 UT_DELETE_ARRAY(log->non_core_fields);
 row_log_block_free(log->tail);
 row_log_block_free(log->head);
 row_merge_file_destroy_low(log->fd);

 if (log->crypt_head) {
  my_large_free(log->crypt_head, log->crypt_head_size);
 }

 if (log->crypt_tail) {
  my_large_free(log->crypt_tail, log->crypt_tail_size);
 }

 mysql_mutex_destroy(&log->mutex);
 ut_free(log);
}

/******************************************************//**
Get the latest transaction ID that has invoked row_log_online_op()
during online creation.
@return latest transaction ID, or 0 if nothing was logged */
trx_id_t
row_log_get_max_trx(
/*================*/
 dict_index_t* index) /*!< in: index, must be locked */
{
 ut_ad(dict_index_get_online_status(index) == ONLINE_INDEX_CREATION);
#ifdef SAFE_MUTEX
 ut_ad(index->lock.have_x()
       || (index->lock.have_s()
    && mysql_mutex_is_owner(&index->online_log->mutex)));
#endif
 return(index->online_log->max_trx);
}

/******************************************************//**
Applies an operation to a secondary index that was being created. */
static MY_ATTRIBUTE((nonnull))
void
row_log_apply_op_low(
/*=================*/
 dict_index_t* index,  /*!< in/out: index */
 row_merge_dup_t*dup,  /*!< in/out: for reporting
duplicate key errors */
 dberr_t* error,  /*!< out: DB_SUCCESS or error code */
 mem_heap_t* offsets_heap, /*!< in/out: memory heap for
allocating offsets; can be emptied */
 bool  has_index_lock, /*!< in: true if holding index->lock
in exclusive mode */
 enum row_op op,  /*!< in: operation being applied */
 trx_id_t trx_id,  /*!< in: transaction identifier */
 const dtuple_t* entry)  /*!< in: row */
{
 mtr_t  mtr{dup->trx};
 btr_cur_t cursor;
 rec_offs* offsets = NULL;

 ut_ad(!dict_index_is_clust(index));

 ut_ad(index->lock.have_x() == has_index_lock);

 ut_ad(!index->is_corrupted());
 ut_ad(trx_id != 0 || op == ROW_OP_DELETE);

 DBUG_LOG("ib_create_index",
   (op == ROW_OP_INSERT ? "insert " : "delete ")
   << (has_index_lock ? "locked index " : "unlocked index ")
   << index->id << ',' << ib::hex(trx_id) << ": "
   << rec_printer(entry).str());

 mtr_start(&mtr);
 index->set_modified(mtr);
 cursor.page_cur.index = index;
 if (has_index_lock) {
  mtr_x_lock_index(index, &mtr);
 }

 /* We perform the pessimistic variant of the operations if we
 already hold index->lock exclusively. First, search the
 record. The operation may already have been performed,
 depending on when the row in the clustered index was
scanned. */
 *error = cursor.search_leaf(entry, PAGE_CUR_LE, has_index_lock
        ? BTR_MODIFY_TREE_ALREADY_LATCHED
        : BTR_MODIFY_LEAF, &mtr);
 if (UNIV_UNLIKELY(*error != DB_SUCCESS)) {
  goto func_exit;
 }

 ut_ad(dict_index_get_n_unique(index) > 0);
 /* This test is somewhat similar to row_ins_must_modify_rec(),
but not identical for unique secondary indexes. */
 if (cursor.low_match >= dict_index_get_n_unique(index)
     && !page_rec_is_infimum(btr_cur_get_rec(&cursor))) {
  /* We have a matching record. */
  bool exists = (cursor.low_match
       == dict_index_get_n_fields(index));
#ifdef UNIV_DEBUG
  rec_t* rec = btr_cur_get_rec(&cursor);
  ut_ad(page_rec_is_user_rec(rec));
  ut_ad(!rec_get_deleted_flag(rec, page_rec_is_comp(rec)));
#endif /* UNIV_DEBUG */

  ut_ad(exists || dict_index_is_unique(index));

  switch (op) {
  case ROW_OP_DELETE:
   if (!exists) {
    /* The existing record matches the
    unique secondary index key, but the
    PRIMARY KEY columns differ. So, this
    exact record does not exist. For
    example, we could detect a duplicate
    key error in some old index before
    logging an ROW_OP_INSERT for our
    index. This ROW_OP_DELETE could have
    been logged for rolling back
TRX_UNDO_INSERT_REC. */
    goto func_exit;
   }

   *error = btr_cur_optimistic_delete(
    &cursor, BTR_CREATE_FLAG, &mtr);

   if (*error != DB_FAIL) {
    break;
   }

   if (!has_index_lock) {
    /* This needs a pessimistic operation.
Lock the index tree exclusively. */
    mtr_commit(&mtr);
    mtr_start(&mtr);
    index->set_modified(mtr);
    *error = cursor.search_leaf(entry, PAGE_CUR_LE,
           BTR_MODIFY_TREE,
           &mtr);
    if (UNIV_UNLIKELY(*error != DB_SUCCESS)) {
     goto func_exit;
    }
    /* No other thread than the current one
    is allowed to modify the index tree.
Thus, the record should still exist. */
    ut_ad(cursor.low_match
          >= dict_index_get_n_fields(index));
    ut_ad(page_rec_is_user_rec(
           btr_cur_get_rec(&cursor)));
   }

   /* As there are no externally stored fields in
   a secondary index record, the parameter
rollback=false will be ignored. */

   btr_cur_pessimistic_delete(
    error, FALSE, &cursor,
    BTR_CREATE_FLAG, false, &mtr);
   break;
  case ROW_OP_INSERT:
   if (exists) {
    /* The record already exists. There
    is nothing to be inserted.
    This could happen when processing
    TRX_UNDO_DEL_MARK_REC in statement
    rollback:

    UPDATE of PRIMARY KEY can lead to
    statement rollback if the updated
    value of the PRIMARY KEY already
    exists. In this case, the UPDATE would
    be mapped to DELETE;INSERT, and we
    only wrote undo log for the DELETE
    part. The duplicate key error would be
    triggered before logging the INSERT
    part.

    Theoretically, we could also get a
    similar situation when a DELETE operation
is blocked by a FOREIGN KEY constraint. */
    goto func_exit;
   }

   if (dtuple_contains_null(entry)) {
    /* The UNIQUE KEY columns match, but
    there is a NULL value in the key, and
NULL!=NULL. */
    goto insert_the_rec;
   }

   goto duplicate;
  }
 } else {
  switch (op) {
   rec_t*  rec;
   big_rec_t* big_rec;
  case ROW_OP_DELETE:
   /* The record does not exist. For example, we
   could detect a duplicate key error in some old
   index before logging an ROW_OP_INSERT for our
   index. This ROW_OP_DELETE could be logged for
rolling back TRX_UNDO_INSERT_REC. */
   goto func_exit;
  case ROW_OP_INSERT:
   if (dict_index_is_unique(index)
       && (cursor.up_match
    >= dict_index_get_n_unique(index)
    || cursor.low_match
    >= dict_index_get_n_unique(index))
       && (!index->n_nullable
    || !dtuple_contains_null(entry))) {
duplicate:
    /* Duplicate key */
    ut_ad(dict_index_is_unique(index));
    row_merge_dup_report(dup, entry->fields);
    *error = DB_DUPLICATE_KEY;
    goto func_exit;
   }
insert_the_rec:
   /* Insert the record. As we are inserting into
   a secondary index, there cannot be externally
stored columns (!big_rec). */
   *error = btr_cur_optimistic_insert(
    BTR_NO_UNDO_LOG_FLAG
    | BTR_NO_LOCKING_FLAG
    | BTR_CREATE_FLAG,
    &cursor, &offsets, &offsets_heap,
    const_cast<dtuple_t*>(entry),
    &rec, &big_rec, 0, NULL, &mtr);
   ut_ad(!big_rec);
   if (*error != DB_FAIL) {
    break;
   }

   if (!has_index_lock) {
    /* This needs a pessimistic operation.
Lock the index tree exclusively. */
    mtr_commit(&mtr);
    mtr_start(&mtr);
    index->set_modified(mtr);
    *error = cursor.search_leaf(entry, PAGE_CUR_LE,
           BTR_MODIFY_TREE,
           &mtr);
    if (*error != DB_SUCCESS) {
     break;
    }
   }

   /* We already determined that the
   record did not exist. No other thread
   than the current one is allowed to
   modify the index tree. Thus, the
record should still not exist. */

   *error = btr_cur_pessimistic_insert(
    BTR_NO_UNDO_LOG_FLAG
    | BTR_NO_LOCKING_FLAG
    | BTR_CREATE_FLAG,
    &cursor, &offsets, &offsets_heap,
    const_cast<dtuple_t*>(entry),
    &rec, &big_rec,
    0, NULL, &mtr);
   ut_ad(!big_rec);
   break;
  }
  mem_heap_empty(offsets_heap);
 }

 if (*error == DB_SUCCESS && trx_id) {
  page_update_max_trx_id(btr_cur_get_block(&cursor),
           btr_cur_get_page_zip(&cursor),
           trx_id, &mtr);
 }

func_exit:
 mtr_commit(&mtr);
}

/******************************************************//**
Applies an operation to a secondary index that was being created.
@return NULL on failure (mrec corruption) or when out of data;
pointer to next record on success */
static MY_ATTRIBUTE((nonnull, warn_unused_result))
const mrec_t*
row_log_apply_op(
/*=============*/
 dict_index_t* index,  /*!< in/out: index */
 row_merge_dup_t*dup,  /*!< in/out: for reporting
duplicate key errors */
 dberr_t* error,  /*!< out: DB_SUCCESS or error code */
 mem_heap_t* offsets_heap, /*!< in/out: memory heap for
allocating offsets; can be emptied */
 mem_heap_t* heap,  /*!< in/out: memory heap for
allocating data tuples */
 bool  has_index_lock, /*!< in: true if holding index->lock
in exclusive mode */
 const mrec_t* mrec,  /*!< in: merge record */
 const mrec_t* mrec_end, /*!< in: end of buffer */
 rec_offs* offsets) /*!< in/out: work area for
rec_init_offsets_temp() */

{
 enum row_op op;
 ulint  extra_size;
 ulint  data_size;
 dtuple_t* entry;
 trx_id_t trx_id;

 /* Online index creation is only used for secondary indexes. */
 ut_ad(!dict_index_is_clust(index));

 ut_ad(index->lock.have_x() == has_index_lock);

 if (index->is_corrupted()) {
  *error = DB_INDEX_CORRUPT;
  return(NULL);
 }

 *error = DB_SUCCESS;

 if (mrec + ROW_LOG_HEADER_SIZE >= mrec_end) {
  return(NULL);
 }

 switch (*mrec) {
 case ROW_OP_INSERT:
  if (ROW_LOG_HEADER_SIZE + DATA_TRX_ID_LEN + mrec >= mrec_end) {
   return(NULL);
  }

  op = static_cast<enum row_op>(*mrec++);
  trx_id = trx_read_trx_id(mrec);
  mrec += DATA_TRX_ID_LEN;
  break;
 case ROW_OP_DELETE:
  op = static_cast<enum row_op>(*mrec++);
  trx_id = 0;
  break;
 default:
corrupted:
  ut_ad(0);
  *error = DB_CORRUPTION;
  return(NULL);
 }

 extra_size = *mrec++;

 ut_ad(mrec < mrec_end);

 if (extra_size >= 0x80) {
  /* Read another byte of extra_size. */

  extra_size = (extra_size & 0x7f) << 8;
  extra_size |= *mrec++;
 }

 mrec += extra_size;

 if (mrec > mrec_end) {
  return(NULL);
 }

 rec_init_offsets_temp(mrec, index, offsets);

 if (rec_offs_any_extern(offsets)) {
  /* There should never be any externally stored fields
  in a secondary index, which is what online index
  creation is used for. Therefore, the log file must be
corrupted. */
  goto corrupted;
 }

 data_size = rec_offs_data_size(offsets);

 mrec += data_size;

 if (mrec > mrec_end) {
  return(NULL);
 }

 entry = row_rec_to_index_entry_low(
  mrec - data_size, index, offsets, heap);
 /* Online index creation is only implemented for secondary
indexes, which never contain off-page columns. */
 ut_ad(dtuple_get_n_ext(entry) == 0);

 row_log_apply_op_low(index, dup, error, offsets_heap,
        has_index_lock, op, trx_id, entry);
 return(mrec);
}

/** Applies operations to a secondary index that was being created.
@param[in] trx transaction (for checking if the operation was
interrupted)
@param[in,out] index index
@param[in,out] dup for reporting duplicate key errors
@param[in,out] stage performance schema accounting object, used by
ALTER TABLE. If not NULL, then stage->inc() will be called for each block
of log that is applied or nullptr when row log applied done by DML
thread.
@return DB_SUCCESS, or error code on failure */
static
dberr_t
row_log_apply_ops(
 const trx_t*  trx,
 dict_index_t*  index,
 row_merge_dup_t* dup,
 ut_stage_alter_t* stage)
{
 dberr_t  error;
 const mrec_t* mrec = NULL;
 const mrec_t* next_mrec;
 const mrec_t* mrec_end= NULL; /* silence bogus warning */
 const mrec_t* next_mrec_end;
 mem_heap_t* offsets_heap;
 mem_heap_t* heap;
 rec_offs* offsets;
 bool  has_index_lock;
 const ulint i = 1 + REC_OFFS_HEADER_SIZE
  + dict_index_get_n_fields(index);

 ut_ad(dict_index_is_online_ddl(index)
       || (index->online_log
    && index->online_status == ONLINE_INDEX_COMPLETE));
 ut_ad(!index->is_committed());
 ut_ad(index->lock.have_x());
 ut_ad(index->online_log);

 MEM_UNDEFINED(&mrec_end, sizeof mrec_end);

 offsets = static_cast<rec_offs*>(ut_malloc_nokey(i * sizeof *offsets));
 rec_offs_set_n_alloc(offsets, i);
 rec_offs_set_n_fields(offsets, dict_index_get_n_fields(index));

 offsets_heap = mem_heap_create(srv_page_size);
 heap = mem_heap_create(srv_page_size);
 has_index_lock = true;

next_block:
 ut_ad(has_index_lock);
 ut_ad(index->lock.have_x());
 ut_ad(index->online_log->head.bytes == 0);

 if (stage) {
  stage->inc(row_log_progress_inc_per_block());
 }

 if (trx_is_interrupted(trx)) {
  goto interrupted;
 }

 error = index->online_log->error;
 if (error != DB_SUCCESS) {
  goto func_exit;
 }

 if (index->is_corrupted()) {
  error = DB_INDEX_CORRUPT;
  goto func_exit;
 }

 if (UNIV_UNLIKELY(index->online_log->head.blocks
     > index->online_log->tail.blocks)) {
unexpected_eof:
  ib::error() << "Unexpected end of temporary file for index "
   << index->name;
corruption:
  error = DB_CORRUPTION;
  goto func_exit;
 }

 if (index->online_log->head.blocks
     == index->online_log->tail.blocks) {
  if (index->online_log->head.blocks) {
#ifdef HAVE_FTRUNCATE
   /* Truncate the file in order to save space. */
   if (index->online_log->fd > 0
       && ftruncate(index->online_log->fd, 0) == -1) {
    ib::error()
     << "\'" << index->name + 1
     << "\' failed with error "
     << errno << ":" << strerror(errno);

    goto corruption;
   }
#endif /* HAVE_FTRUNCATE */
   index->online_log->head.blocks
    = index->online_log->tail.blocks = 0;
  }

  next_mrec = index->online_log->tail.block;
  next_mrec_end = next_mrec + index->online_log->tail.bytes;

  if (next_mrec_end == next_mrec) {
   /* End of log reached. */
all_done:
   ut_ad(has_index_lock);
   ut_ad(index->online_log->head.blocks == 0);
   ut_ad(index->online_log->tail.blocks == 0);
   index->online_log->tail.bytes = 0;
   index->online_log->head.bytes = 0;
   error = DB_SUCCESS;
   goto func_exit;
  }
 } else {
  os_offset_t ofs = static_cast<os_offset_t>(
   index->online_log->head.blocks)
   * srv_sort_buf_size;
  ut_ad(has_index_lock);
  has_index_lock = false;
  index->lock.x_unlock();

  log_free_check();

  if (!row_log_block_allocate(index->online_log->head)) {
   error = DB_OUT_OF_MEMORY;
   goto func_exit;
  }

  byte* buf = index->online_log->head.block;

  if (DB_SUCCESS
      != os_file_read(IORequestRead, index->online_log->fd,
        buf, ofs, srv_sort_buf_size, nullptr)) {
   ib::error()
    << "Unable to read temporary file"
    " for index " << index->name;
   goto corruption;
  }

  if (srv_encrypt_log) {
   if (!log_tmp_block_decrypt(
        buf, srv_sort_buf_size,
        index->online_log->crypt_head, ofs)) {
    error = DB_DECRYPTION_FAILED;
    goto func_exit;
   }

   srv_stats.n_rowlog_blocks_decrypted.inc();
   memcpy(buf, index->online_log->crypt_head, srv_sort_buf_size);
  }

#ifdef POSIX_FADV_DONTNEED
  /* Each block is read exactly once.  Free up the file cache. */
  posix_fadvise(index->online_log->fd,
         ofs, srv_sort_buf_size, POSIX_FADV_DONTNEED);
#endif /* POSIX_FADV_DONTNEED */

  next_mrec = index->online_log->head.block;
  next_mrec_end = next_mrec + srv_sort_buf_size;
 }

 if (mrec) {
  /* A partial record was read from the previous block.
  Copy the temporary buffer full, as we do not know the
  length of the record. Parse subsequent records from
  the bigger buffer index->online_log->head.block
or index->online_log->tail.block. */

  ut_ad(mrec == index->online_log->head.buf);
  ut_ad(mrec_end > mrec);
  ut_ad(mrec_end < (&index->online_log->head.buf)[1]);

  memcpy((mrec_t*) mrec_end, next_mrec,
         ulint((&index->online_log->head.buf)[1] - mrec_end));
  mrec = row_log_apply_op(
   index, dup, &error, offsets_heap, heap,
   has_index_lock, index->online_log->head.buf,
   (&index->online_log->head.buf)[1], offsets);
  if (error != DB_SUCCESS) {
   goto func_exit;
  } else if (UNIV_UNLIKELY(mrec == NULL)) {
   /* The record was not reassembled properly. */
   goto corruption;
  }
  /* The record was previously found out to be
  truncated. Now that the parse buffer was extended,
it should proceed beyond the old end of the buffer. */
  ut_a(mrec > mrec_end);

  index->online_log->head.bytes = ulint(mrec - mrec_end);
  next_mrec += index->online_log->head.bytes;
 }

 ut_ad(next_mrec <= next_mrec_end);
 /* The following loop must not be parsing the temporary
buffer, but head.block or tail.block. */

 /* mrec!=NULL means that the next record starts from the
middle of the block */
 ut_ad((mrec == NULL) == (index->online_log->head.bytes == 0));

#ifdef UNIV_DEBUG
 if (next_mrec_end - srv_sort_buf_size
            == index->online_log->head.block) {
  /* If tail.bytes == 0, next_mrec_end can also be at
the end of tail.block. */
  if (index->online_log->tail.bytes == 0) {
   ut_ad(next_mrec == next_mrec_end);
   ut_ad(index->online_log->tail.blocks == 0);
   ut_ad(index->online_log->head.blocks == 0);
   ut_ad(index->online_log->head.bytes == 0);
  } else {
   ut_ad(next_mrec == index->online_log->head.block
         + index->online_log->head.bytes);
   ut_ad(index->online_log->tail.blocks
         > index->online_log->head.blocks);
  }
 } else if (next_mrec_end - index->online_log->tail.bytes
                   == index->online_log->tail.block) {
  ut_ad(next_mrec == index->online_log->tail.block
        + index->online_log->head.bytes);
  ut_ad(index->online_log->tail.blocks == 0);
  ut_ad(index->online_log->head.blocks == 0);
  ut_ad(index->online_log->head.bytes
        <= index->online_log->tail.bytes);
 } else {
  ut_error;
 }
#endif /* UNIV_DEBUG */

 mrec_end = next_mrec_end;

 while (!trx_is_interrupted(trx)) {
  mrec = next_mrec;
  ut_ad(mrec < mrec_end);

  if (!has_index_lock) {
   /* We are applying operations from a different
   block than the one that is being written to.
   We do not hold index->lock in order to
   allow other threads to concurrently buffer
modifications. */
   ut_ad(mrec >= index->online_log->head.block);
   ut_ad(mrec_end == index->online_log->head.block
         + srv_sort_buf_size);
   ut_ad(index->online_log->head.bytes
         < srv_sort_buf_size);

   /* Take the opportunity to do a redo log
checkpoint if needed. */
   log_free_check();
  } else {
   /* We are applying operations from the last block.
   Do not allow other threads to buffer anything,
so that we can finally catch up and synchronize. */
   ut_ad(index->online_log->head.blocks == 0);
   ut_ad(index->online_log->tail.blocks == 0);
   ut_ad(mrec_end == index->online_log->tail.block
         + index->online_log->tail.bytes);
   ut_ad(mrec >= index->online_log->tail.block);
  }

  next_mrec = row_log_apply_op(
   index, dup, &error, offsets_heap, heap,
   has_index_lock, mrec, mrec_end, offsets);

  if (error != DB_SUCCESS) {
   goto func_exit;
  } else if (next_mrec == next_mrec_end) {
   /* The record happened to end on a block boundary.
Do we have more blocks left? */
   if (has_index_lock) {
    /* The index will be locked while
applying the last block. */
    goto all_done;
   }

   mrec = NULL;
process_next_block:
   index->lock.x_lock(SRW_LOCK_CALL);
   has_index_lock = true;

   index->online_log->head.bytes = 0;
   index->online_log->head.blocks++;
   goto next_block;
  } else if (next_mrec != NULL) {
   ut_ad(next_mrec < next_mrec_end);
   index->online_log->head.bytes
    += ulint(next_mrec - mrec);
  } else if (has_index_lock) {
   /* When mrec is within tail.block, it should
   be a complete record, because we are holding
index->lock and thus excluding the writer. */
   ut_ad(index->online_log->tail.blocks == 0);
   ut_ad(mrec_end == index->online_log->tail.block
         + index->online_log->tail.bytes);
   ut_ad(0);
   goto unexpected_eof;
  } else {
   const size_t s = size_t(mrec_end - mrec);
   memcpy(index->online_log->head.buf, mrec, s);
   mrec = index->online_log->head.buf;
   mrec_end = mrec + s;
   goto process_next_block;
  }
 }

interrupted:
 error = DB_INTERRUPTED;
func_exit:
 if (!has_index_lock) {
  index->lock.x_lock(SRW_LOCK_CALL);
 }

 switch (error) {
 case DB_SUCCESS:
  break;
 case DB_INDEX_CORRUPT:
  if (((os_offset_t) index->online_log->tail.blocks + 1)
      * srv_sort_buf_size >= srv_online_max_size) {
   /* The log file grew too big. */
   error = DB_ONLINE_LOG_TOO_BIG;
  }
  /* fall through */
 default:
  index->type |= DICT_CORRUPT;
 }

 mem_heap_free(heap);
 mem_heap_free(offsets_heap);
 row_log_block_free(index->online_log->head);
 ut_free(offsets);
 return(error);
}

/** Apply the row log to the index upon completing index creation.
@param[in,out] trx transaction (for checking if the operation was
interrupted)
@param[in,out] index secondary index
@param[in,out] table MySQL table (for reporting duplicates)
@param[in,out] stage performance schema accounting object, used by
ALTER TABLE. stage->begin_phase_log_index() will be called initially and then
stage->inc() will be called for each block of log that is applied or nullptr
when row log has been applied by DML thread.
@return DB_SUCCESS, or error code on failure */
dberr_t
row_log_apply(
 trx_t*   trx,
 dict_index_t*  index,
 struct TABLE*  table,
 ut_stage_alter_t* stage)
{
 dberr_t  error;
 row_merge_dup_t dup = { index, trx, table, nullptr, 0 };
 DBUG_ENTER("row_log_apply");

 ut_ad(dict_index_is_online_ddl(index)
       || (index->online_log
    && index->online_status == ONLINE_INDEX_COMPLETE));
 ut_ad(!dict_index_is_clust(index));

 if (stage) {
  stage->begin_phase_log_index();
 }

 log_free_check();

 index->lock.x_lock(SRW_LOCK_CALL);

 if (index->online_log && !index->table->corrupted) {
  error = row_log_apply_ops(trx, index, &dup, stage);
 } else {
  error = DB_SUCCESS;
 }

 if (error != DB_SUCCESS) {
  ut_ad(index->table->space);
  index->type |= DICT_CORRUPT;
  index->table->drop_aborted = TRUE;

  dict_index_set_online_status(index, ONLINE_INDEX_ABORTED);
 } else if (stage) {
  /* Mark the index as completed only when it is
being called by DDL thread */
  ut_ad(dup.n_dup == 0);
  dict_index_set_online_status(index, ONLINE_INDEX_COMPLETE);
 }

 index->lock.x_unlock();

 DBUG_RETURN(error);
}

unsigned row_log_get_n_core_fields(const dict_index_t *index)
{
  ut_ad(index->online_log);
  return index->online_log->n_core_fields;
}

dberr_t row_log_get_error(const dict_index_t *index)
{
  ut_ad(index->online_log);
  return index->online_log->error;
}

dberr_t dict_table_t::clear(que_thr_t *thr)
{
  dberr_t err= DB_SUCCESS;
  for (dict_index_t *index= UT_LIST_GET_FIRST(indexes); index;
       index= UT_LIST_GET_NEXT(indexes, index))
  {
    if (index->type & DICT_FTS)
      continue;

    switch (dict_index_get_online_status(index)) {
    case ONLINE_INDEX_ABORTED:
    case ONLINE_INDEX_ABORTED_DROPPED:
      continue;
    case ONLINE_INDEX_COMPLETE:
      break;
    case ONLINE_INDEX_CREATION:
      ut_ad("invalid type" == 0);
      MY_ASSERT_UNREACHABLE();
      break;
    }
    if (dberr_t err_index= index->clear(thr))
      err= err_index;
  }
  return err;
}

inline bool UndorecApplier::is_same(roll_ptr_t roll_ptr) const
{
  return uint16_t(roll_ptr) == offset &&
    uint32_t(roll_ptr >> 16) == page_id.page_no();
}

const rec_t *
UndorecApplier::get_old_rec(const dtuple_t &tuple, dict_index_t *index,
                            const rec_t **clust_rec, rec_offs **offsets)
{
  ut_ad(index->is_primary());
  btr_pcur_t pcur;

  bool found= row_search_on_row_ref(&pcur, BTR_MODIFY_LEAF,
                                    index->table, &tuple, &mtr);
  ut_a(found);
  *clust_rec= btr_pcur_get_rec(&pcur);

  ulint len= 0;
  rec_t *prev_version;
  const rec_t *version= *clust_rec;
  do
  {
    *offsets= rec_get_offsets(version, index, *offsets,
                              index->n_core_fields, ULINT_UNDEFINED,
                              &heap);
    roll_ptr_t roll_ptr= trx_read_roll_ptr(
      rec_get_nth_field(version, *offsets, index->db_roll_ptr(), &len));
    ut_ad(len == DATA_ROLL_PTR_LEN);
    if (is_same(roll_ptr))
      return version;
    trx_undo_prev_version_build(version, index, *offsets, heap, &prev_version,
                                &mtr, 0, nullptr, nullptr);
    version= prev_version;
  }
  while (version);

  return nullptr;
}

/** Clear out all online log of other online indexes after
encountering the error during row_log_apply() in DML thread
@param table table which does online DDL */
static void row_log_mark_other_online_index_abort(dict_table_t *table)
{
  dict_index_t *clust_index= dict_table_get_first_index(table);
  for (dict_index_t *index= dict_table_get_next_index(clust_index);
       index; index= dict_table_get_next_index(index))
  {
    if (index->online_log &&
        index->online_status <= ONLINE_INDEX_CREATION &&
        !index->is_corrupted())
    {
      index->lock.x_lock(SRW_LOCK_CALL);
      row_log_abort_sec(index);
      index->type|= DICT_CORRUPT;
      index->lock.x_unlock();
      MONITOR_ATOMIC_INC(MONITOR_BACKGROUND_DROP_INDEX);
    }
  }

  clust_index->lock.x_lock(SRW_LOCK_CALL);
  clust_index->online_log= nullptr;
  clust_index->lock.x_unlock();
  table->drop_aborted= TRUE;
}

void dtype_t::assign(const dict_col_t &col)
{
  prtype= col.prtype;
  mtype= col.mtype;
  len= col.len;
  mbminlen= col.mbminlen;
  mbmaxlen= col.mbmaxlen;
}

inline void dtuple_t::copy_field_types(const dict_index_t &index)
{
  ut_ad(index.n_fields == n_fields);
  if (UNIV_LIKELY_NULL(index.change_col_info))
    for (ulint i= 0; i < n_fields; i++)
      fields[i].type.assign(*index.fields[i].col);
}

void UndorecApplier::log_insert(const dtuple_t &tuple,
                                dict_index_t *clust_index)
{
  DEBUG_SYNC_C("row_log_insert_handle");
  ut_ad(clust_index->is_primary());
  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets= offsets_;

  rec_offs_init(offsets_);
  mtr.start();
  const rec_t *rec;
  const rec_t *match_rec= get_old_rec(tuple, clust_index, &rec, &offsets);
  if (!match_rec)
  {
    mtr.commit();
    return;
  }
  const rec_t *copy_rec= match_rec;
  if (match_rec == rec)
  {
    copy_rec= rec_copy(mem_heap_alloc(
      heap, rec_offs_size(offsets)), match_rec, offsets);
    rec_offs_make_valid(copy_rec, clust_index, true, offsets);
  }
  mtr.commit();

  dict_table_t *table= clust_index->table;
  clust_index->lock.s_lock(SRW_LOCK_CALL);
  if (clust_index->online_log &&
      !clust_index->online_log_is_dummy() &&
      clust_index->online_status <= ONLINE_INDEX_CREATION)
  {
    row_log_table_insert(copy_rec, clust_index, offsets);
    clust_index->lock.s_unlock();
  }
  else
  {
    clust_index->lock.s_unlock();
    row_ext_t *ext;
    dtuple_t *row= row_build(ROW_COPY_POINTERS, clust_index,
      copy_rec, offsets, table, nullptr, nullptr, &ext, heap);

    if (table->n_v_cols)
    {
      /* Update the row with virtual column values present
      in the undo log or update vector */
      if (type == TRX_UNDO_UPD_DEL_REC)
        row_upd_replace_vcol(row, table, update, false, nullptr,
                             (cmpl_info & UPD_NODE_NO_ORD_CHANGE)
                             ? nullptr : undo_rec);
      else
        trx_undo_read_v_cols(table, undo_rec, row, false);
    }

    bool success= true;
    for (dict_index_t *index= clust_index;
         (index= dict_table_get_next_index(index)) != nullptr; )
    {
      index->lock.s_lock(SRW_LOCK_CALL);
      if (index->online_log &&
          index->online_status <= ONLINE_INDEX_CREATION &&
          !index->is_corrupted())
      {
        dtuple_t *entry= row_build_index_entry_low(row, ext, index,
                                                   heap, ROW_BUILD_NORMAL);
        entry->copy_field_types(*index);
 success= row_log_online_op(index, entry, mtr.trx, trx_id);
      }

      index->lock.s_unlock();
      if (!success)
      {
        row_log_mark_other_online_index_abort(index->table);
        return;
      }
    }
  }
}

void UndorecApplier::log_update(const dtuple_t &tuple,
                                dict_index_t *clust_index)
{
  rec_offs offsets_[REC_OFFS_NORMAL_SIZE];
  rec_offs offsets2_[REC_OFFS_NORMAL_SIZE];
  rec_offs *offsets= offsets_;
  rec_offs *prev_offsets= offsets2_;

  rec_offs_init(offsets_);
  rec_offs_init(offsets2_);

  dict_table_t *table= clust_index->table;

  clust_index->lock.s_lock(SRW_LOCK_CALL);
  bool table_rebuild=
    (clust_index->online_log
     && !clust_index->online_log_is_dummy()
     && clust_index->online_status <= ONLINE_INDEX_CREATION);
  clust_index->lock.s_unlock();

  mtr.start();
  const rec_t *rec;
  rec_t *prev_version;
  bool is_update= (type == TRX_UNDO_UPD_EXIST_REC);
  const rec_t *match_rec= get_old_rec(tuple, clust_index, &rec, &offsets);
  if (!match_rec)
  {
    mtr.commit();
    return;
  }

  if (table_rebuild)
  {
    const rec_t *copy_rec= match_rec;
    if (match_rec == rec)
      copy_rec= rec_copy(mem_heap_alloc(
        heap, rec_offs_size(offsets)), match_rec, offsets);
    trx_undo_prev_version_build(match_rec, clust_index, offsets, heap,
                                &prev_version, &mtr, 0, nullptr, nullptr);

    prev_offsets= rec_get_offsets(prev_version, clust_index, prev_offsets,
                                  clust_index->n_core_fields,
                                  ULINT_UNDEFINED, &heap);
    rec_offs_make_valid(copy_rec, clust_index, true, offsets);
    mtr.commit();

    clust_index->lock.s_lock(SRW_LOCK_CALL);
    /* Recheck whether clustered index online log has been cleared */
    if (clust_index->online_log)
    {
      if (is_update)
      {
        const dtuple_t *rebuilt_old_pk= row_log_table_get_pk(
          prev_version, clust_index, prev_offsets, nullptr, &heap);
        row_log_table_update(copy_rec, clust_index, offsets, rebuilt_old_pk);
      }
      else
        row_log_table_delete(prev_version, clust_index, prev_offsets, nullptr);
    }
    clust_index->lock.s_unlock();
    return;
  }

  dtuple_t *row= nullptr;
  row_ext_t *new_ext;
  if (match_rec != rec)
    row= row_build(ROW_COPY_POINTERS, clust_index, match_rec, offsets,
                   clust_index->table, NULL, NULL, &new_ext, heap);
  else
    row= row_build(ROW_COPY_DATA, clust_index, rec, offsets,
                   clust_index->table, NULL, NULL, &new_ext, heap);
  mtr.commit();
  row_ext_t *old_ext;
  dtuple_t *old_row= nullptr;
  if (!(this->cmpl_info & UPD_NODE_NO_ORD_CHANGE))
  {
    for (ulint i = 0; i < dict_table_get_n_v_cols(table); i++)
     dfield_get_type(dtuple_get_nth_v_field(row, i))->mtype = DATA_MISSING;
  }

  if (table->n_v_cols)
    row_upd_replace_vcol(row, table, update, false, nullptr,
                         (cmpl_info & UPD_NODE_NO_ORD_CHANGE)
                         ? nullptr : undo_rec);

  if (is_update)
  {
    old_row= dtuple_copy(row, heap);
    row_upd_replace(old_row, &old_ext, clust_index, update, heap);
  }

  bool success= true;
  dict_index_t *index= dict_table_get_next_index(clust_index);
  while (index)
  {
    index->lock.s_lock(SRW_LOCK_CALL);
    if (index->online_log &&
        index->online_status <= ONLINE_INDEX_CREATION &&
        !index->is_corrupted())
    {
      if (is_update)
      {
        /* Ignore the index if the update doesn't affect the index */
        if (!row_upd_changes_ord_field_binary(index, update,
                                              nullptr,
                                              row, new_ext))
          goto next_index;
        dtuple_t *old_entry= row_build_index_entry_low(
          old_row, old_ext, index, heap, ROW_BUILD_NORMAL);

        old_entry->copy_field_types(*index);

 success= row_log_online_op(index, old_entry, mtr.trx, 0);

 dtuple_t *new_entry= row_build_index_entry_low(
          row, new_ext, index, heap, ROW_BUILD_NORMAL);

        new_entry->copy_field_types(*index);

 if (success)
   success= row_log_online_op(index, new_entry, mtr.trx, trx_id);
      }
      else
      {
        dtuple_t *old_entry= row_build_index_entry_low(
          row, new_ext, index, heap, ROW_BUILD_NORMAL);

        old_entry->copy_field_types(*index);

        success= row_log_online_op(index, old_entry, mtr.trx, 0);
      }
    }
next_index:
    index->lock.s_unlock();
    if (!success)
    {
      row_log_mark_other_online_index_abort(index->table);
      return;
    }
    index= dict_table_get_next_index(index);
  }
}