/*
   Copyright (c) 2000, 2018, Oracle and/or its affiliates.
   Copyright (c) 2009, 2022, MariaDB

   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 St, Fifth Floor, Boston, MA 02110-1335  USA */


#include "mariadb.h"
#include "sql_priv.h"
#include "handler.h"
#ifndef MYSQL_CLIENT
#include "unireg.h"
#include "log_event.h"
#include "sql_base.h"                           // close_thread_tables
#include "sql_cache.h"                       // QUERY_CACHE_FLAGS_SIZE
#include "sql_locale.h" // MY_LOCALE, my_locale_by_number, my_locale_en_US
#include "key.h"        // key_copy
#include "lock.h"       // mysql_unlock_tables
#include "sql_parse.h"  // mysql_test_parse_for_slave
#include "tztime.h"     // struct Time_zone
#include "sql_load.h"   // mysql_load
#include "sql_db.h"     // load_db_opt_by_name
#include "slave.h"
#include "rpl_rli.h"
#include "rpl_mi.h"
#include "rpl_filter.h"
#include "rpl_record.h"
#include "transaction.h"
#include <my_dir.h>
#include "sql_show.h"    // append_identifier
#include <mysql/psi/mysql_statement.h>
#include <strfunc.h>
#include "compat56.h"
#include "sql_insert.h"
#ifdef WITH_WSREP
#include "wsrep_mysqld.h"
#endif /* WITH_WSREP */
#else
#include "mysqld_error.h"
#endif /* MYSQL_CLIENT */

#include <my_bitmap.h>
#include "rpl_utility.h"
#include "rpl_constants.h"
#include "sql_digest.h"
#include "zlib.h"
#include "myisampack.h"
#include <algorithm>

#define my_b_write_string(A, B) my_b_write((A), (uchar*)(B), (uint) (sizeof(B) - 1))

#ifndef _AIX
PSI_memory_key key_memory_log_event;
#endif
PSI_memory_key key_memory_Incident_log_event_message;

/**
  BINLOG_CHECKSUM variable.
*/
const char *binlog_checksum_type_names[]= {
  "NONE",
  "CRC32",
  NullS
};

unsigned int binlog_checksum_type_length[]= {
  sizeof("NONE") - 1,
  sizeof("CRC32") - 1,
  0
};

TYPELIB binlog_checksum_typelib= CREATE_TYPELIB_FOR(binlog_checksum_type_names);

#define FLAGSTR(V,F) ((V)&(F)?#F" ":"")

/*
  Size of buffer for printing a double in format %.<PREC>g

  optional '-' + optional zero + '.'  + PREC digits + 'e' + sign +
  exponent digits + '\0'
*/
#define FMT_G_BUFSIZE(PREC) (3 + (PREC) + 5 + 1)

/* 
   replication event checksum is introduced in the following "checksum-home" version.
   The checksum-aware servers extract FD's version to decide whether the FD event
   carries checksum info.

   TODO: correct the constant when it has been determined 
   (which main tree to push and when) 
*/
const Version checksum_version_split_mysql(5, 6, 1);
const Version checksum_version_split_mariadb(5, 3, 0);

// First MySQL version with fraction seconds
const Version fsp_version_split_mysql(5, 6, 0);

/*
  Cache that will automatically be written to a dedicated file on
  destruction.

  DESCRIPTION

 */
class Write_on_release_cache
{
public:
  enum flag
  {
    FLUSH_F
  };

  typedef unsigned short flag_set;

  /*
    Constructor.

    SYNOPSIS
      Write_on_release_cache
      cache  Pointer to cache to use
      file   File to write cache to upon destruction
      flags  Flags for the cache

    DESCRIPTION
      Cache common parameters and ensure common flush_data() code
      on successful copy of the cache, the cache will be reinited as a
      WRITE_CACHE.

      Currently, a pointer to the cache is provided in the
      constructor, but it would be possible to create a subclass
      holding the IO_CACHE itself.
   */
  Write_on_release_cache(IO_CACHE *cache, FILE *file, flag_set flags= 0, Log_event *ev= NULL)
    : m_cache(cache), m_file(file), m_flags(flags), m_ev(ev)
  {
    reinit_io_cache(m_cache, WRITE_CACHE, 0L, FALSE, TRUE);
  }

  ~Write_on_release_cache() = default;

  bool flush_data()
  {
#ifdef MYSQL_CLIENT
    if (m_ev == NULL)
    {
      if (copy_event_cache_to_file_and_reinit(m_cache, m_file))
        return 1;
      if ((m_flags & FLUSH_F) && fflush(m_file))
        return 1;
    }
    else // if m_ev<>NULL, then storing the output in output_buf
    {
      LEX_STRING tmp_str;
      bool res;
      if (copy_event_cache_to_string_and_reinit(m_cache, &tmp_str))
        return 1;
      /* use 2 argument append as tmp_str is not \0 terminated */
      res= m_ev->output_buf.append(tmp_str.str, tmp_str.length);
      my_free(tmp_str.str);
      return res ? res : 0;
    }
#else /* MySQL_SERVER */
    if (copy_event_cache_to_file_and_reinit(m_cache, m_file))
      return 1;
    if ((m_flags & FLUSH_F) && fflush(m_file))
      return 1;
#endif
    return 0;
  }

  /*
    Return a pointer to the internal IO_CACHE.

    SYNOPSIS
      operator&()

    DESCRIPTION

      Function to return a pointer to the internal cache, so that the
      object can be treated as a IO_CACHE and used with the my_b_*
      IO_CACHE functions

    RETURN VALUE
      A pointer to the internal IO_CACHE.
   */
  IO_CACHE *operator&()
  {
    return m_cache;
  }

private:
  // Hidden, to prevent usage.
  Write_on_release_cache(Write_on_release_cache const&);

  IO_CACHE *m_cache;
  FILE *m_file;
  flag_set m_flags;
  Log_event *m_ev; // Used for Flashback
};

#if !defined DBUG_OFF && defined DBUG_TRACE
#define DBUG_DUMP_EVENT_BUF(B,L)                                         \
  do {                                                                   \
    const uchar *_buf=(uchar*)(B);                                       \
    size_t _len=(L);                                                     \
    if (_len >= LOG_EVENT_MINIMAL_HEADER_LEN)                            \
    {                                                                    \
      DBUG_PRINT("data", ("header: timestamp:%u type:%u server_id:%u len:%u log_pos:%u flags:%u",  \
                          uint4korr(_buf), _buf[EVENT_TYPE_OFFSET],      \
                          uint4korr(_buf+SERVER_ID_OFFSET),              \
                          uint4korr(_buf+EVENT_LEN_OFFSET),              \
                          uint4korr(_buf+LOG_POS_OFFSET),                \
                          uint4korr(_buf+FLAGS_OFFSET)));                \
      DBUG_DUMP("data", _buf+LOG_EVENT_MINIMAL_HEADER_LEN,               \
                _len-LOG_EVENT_MINIMAL_HEADER_LEN);                      \
    }                                                                    \
    else                                                                 \
      DBUG_DUMP("data", _buf, _len);                                     \
  } while(0)
#else
#define DBUG_DUMP_EVENT_BUF(B,L) do { } while(0)
#endif

/*
  read_str()
*/

static inline bool read_str(const uchar **buf, const uchar *buf_end,
                            const char **str, uint8 *len)
{
  if (*buf + ((uint) **buf) >= buf_end)
    return 1;
  *len= (uint8) **buf;
  *str= (char*) (*buf)+1;
  (*buf)+= (uint) *len+1;
  return 0;
}


/**
  Transforms a string into "" or its expression in X'HHHH' form.
*/

char *str_to_hex(char *to, const uchar *from, size_t len)
{
  if (len)
  {
    *to++= 'X';
    *to++= '\'';
    to= octet2hex(to, from, len);
    *to++= '\'';
    *to= '\0';
  }
  else
    to= strmov(to, "\"\"");
  return to;                               // pointer to end 0 of 'to'
}

#define BINLOG_COMPRESSED_HEADER_LEN 1
#define BINLOG_COMPRESSED_ORIGINAL_LENGTH_MAX_BYTES 4
/**
  Compressed Record
    Record Header: 1 Byte
             7 Bit: Always 1, mean compressed;
           4-6 Bit: Compressed algorithm - Always 0, means zlib
                    It maybe support other compression algorithm in the future.
           0-3 Bit: Bytes of "Record Original Length"
    Record Original Length: 1-4 Bytes
    Compressed Buf:
*/

/**
  Get the length of compress content.
*/

uint32 binlog_get_compress_len(uint32 len)
{
    /* 5 for the begin content, 1 reserved for a '\0'*/
    return ALIGN_SIZE((BINLOG_COMPRESSED_HEADER_LEN + BINLOG_COMPRESSED_ORIGINAL_LENGTH_MAX_BYTES) 
                        + compressBound(len) + 1);
}

/**
   Compress buf from 'src' to 'dst'.

   Note: 1) Then the caller should guarantee the length of 'dst', which
      can be got by binlog_get_uncompress_len, is enough to hold
      the content uncompressed.
         2) The 'comlen' should stored the length of 'dst', and it will
      be set as the size of compressed content after return.

   return zero if successful, others otherwise.
*/
int binlog_buf_compress(const uchar *src, uchar *dst, uint32 len, uint32 *comlen)
{
  uchar lenlen;
  if (len & 0xFF000000)
  {
    dst[1]= uchar(len >> 24);
    dst[2]= uchar(len >> 16);
    dst[3]= uchar(len >> 8);
    dst[4]= uchar(len);
    lenlen= 4;
  }
  else if (len & 0x00FF0000)
  {
    dst[1]= uchar(len >> 16);
    dst[2]= uchar(len >> 8);
    dst[3]= uchar(len);
    lenlen= 3;
  }
  else if (len & 0x0000FF00)
  {
    dst[1]= uchar(len >> 8);
    dst[2]= uchar(len);
    lenlen= 2;
  }
  else 
  {
    dst[1]= uchar(len);
    lenlen= 1;
  }

  /*
    Write the length of an event no replica can uncompress, so that a
    replica's handling of one can be tested. Reaching a value this large
    takes four length bytes, and the encoding above spends four only on
    content of 16MB or more, so the injection sets the encoding as well
    as the value. The compressed content still goes where those four
    bytes place it, leaving the event well formed apart from the length
    it claims.
  */
  DBUG_EXECUTE_IF("binlog_compress_corrupt_len",
                  {
                    dst[1]= 0xFF;
                    dst[2]= 0xFF;
                    dst[3]= 0xFF;
                    dst[4]= 0xFC;
                    lenlen= 4;
                  });

  dst[0]= 0x80 | (lenlen & 0x07);

  uLongf tmplen= (uLongf)*comlen - BINLOG_COMPRESSED_HEADER_LEN - lenlen - 1;
  if (compress((Bytef *)dst + BINLOG_COMPRESSED_HEADER_LEN + lenlen, &tmplen,
               (const Bytef *)src, (uLongf)len) != Z_OK)
  {
    return 1;
  }
  *comlen= (uint32)tmplen + BINLOG_COMPRESSED_HEADER_LEN + lenlen;
  return 0;
}

/**
   Convert a query_compressed_log_event to query_log_event
   from 'src' to 'dst', the size after compression stored in 'newlen'.

   @Note:
      1) The caller should call my_free to release 'dst' if *is_malloc is
         returned as true.
      2) If *is_malloc is returned as false, then 'dst' reuses the passed-in
         'buf'.

   return zero if successful, otherwise the error code the caller reports.
*/

int
query_event_uncompress(const Format_description_log_event *description_event,
                       bool contain_checksum, const uchar *src, ulong src_len,
                       uchar* buf, ulong buf_size, bool* is_malloc, uchar **dst,
                       ulong *newlen)
{
  ulong len= uint4korr(src + EVENT_LEN_OFFSET);
  const uchar *tmp= src;
  const uchar *end= src + len;
  uchar *new_dst;

  // bad event
  if (unlikely(src_len < len))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  DBUG_ASSERT((uchar)src[EVENT_TYPE_OFFSET] == QUERY_COMPRESSED_EVENT);

  uint8 common_header_len= description_event->common_header_len;
  uint8 post_header_len=
    description_event->post_header_len[QUERY_COMPRESSED_EVENT-1];

  *is_malloc= false;

  tmp+= common_header_len;
  // bad event
  if (unlikely(end <= tmp))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  uint db_len= (uint)tmp[Q_DB_LEN_OFFSET];
  uint16 status_vars_len= uint2korr(tmp + Q_STATUS_VARS_LEN_OFFSET);

  tmp+= post_header_len + status_vars_len + db_len + 1;
  // bad event
  if (unlikely(end <= tmp))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  int32 comp_len= (int32)(len - (tmp - src) -
                          (contain_checksum ? BINLOG_CHECKSUM_LEN : 0));
  uint32 un_len=  binlog_get_uncompress_len(tmp);

  // bad event 
  if (unlikely(comp_len < 0 || un_len == 0))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  if (unlikely(un_len > MAX_MAX_ALLOWED_PACKET))
    return ER_TOO_BIG_FOR_UNCOMPRESS;

  *newlen= (ulong)(tmp - src) + un_len;
  if (contain_checksum)
    *newlen+= BINLOG_CHECKSUM_LEN;

  size_t alloc_size= ALIGN_SIZE(*newlen);
  
  if (alloc_size <= buf_size) 
    new_dst= buf;
  else 
  {
    new_dst= (uchar *) my_malloc(PSI_INSTRUMENT_ME, alloc_size, MYF(MY_WME));
    if (!new_dst)
      return ER_BINLOG_UNCOMPRESS_ERROR;
    *is_malloc= true;
  }

  /* copy the head*/
  memcpy(new_dst, src , tmp - src);
  if (binlog_buf_uncompress(tmp, new_dst + (tmp - src), comp_len, &un_len))
  {
    if (*is_malloc)
    {
      *is_malloc= false;
      my_free(new_dst);
    }
    return ER_BINLOG_UNCOMPRESS_ERROR;
  }

  new_dst[EVENT_TYPE_OFFSET]= QUERY_EVENT;
  int4store(new_dst + EVENT_LEN_OFFSET, *newlen);
  if (contain_checksum)
  {
    ulong clear_len= *newlen - BINLOG_CHECKSUM_LEN;
    int4store(new_dst + clear_len,
              my_checksum(0L, (uchar *)new_dst, clear_len));
  }
  *dst= new_dst;
  return 0;
}

int
row_log_event_uncompress(const Format_description_log_event *description_event,
                         bool contain_checksum, const uchar *src, ulong src_len,
                         uchar* buf, ulong buf_size, bool* is_malloc,
                         uchar **dst, ulong *newlen)
{
  Log_event_type type= (Log_event_type)(uchar)src[EVENT_TYPE_OFFSET];
  ulong len= uint4korr(src + EVENT_LEN_OFFSET);
  const uchar *tmp= src;
  uchar *new_dst= NULL;
  const uchar *end= tmp + len;

  if (unlikely(src_len < len))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  DBUG_ASSERT(LOG_EVENT_IS_ROW_COMPRESSED(type));

  uint8 common_header_len= description_event->common_header_len;
  uint8 post_header_len= description_event->post_header_len[type-1];

  tmp+= common_header_len + ROWS_HEADER_LEN_V1;
  if (post_header_len == ROWS_HEADER_LEN_V2)
  {
    /*
      Have variable length header, check length,
      which includes length bytes
    */

    if (unlikely(end - tmp <= 2))
      return ER_BINLOG_UNCOMPRESS_ERROR;

    uint16 var_header_len= uint2korr(tmp);
    DBUG_ASSERT(var_header_len >= 2);

    /* skip over var-len header, extracting 'chunks' */
    tmp+= var_header_len;

    /* get the uncompressed event type */
    type=
      (Log_event_type)(type - WRITE_ROWS_COMPRESSED_EVENT + WRITE_ROWS_EVENT);
  }
  else 
  {
    /* get the uncompressed event type */
    type= (Log_event_type)
      (type - WRITE_ROWS_COMPRESSED_EVENT_V1 + WRITE_ROWS_EVENT_V1);
  }

  if (unlikely(end <= tmp))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  ulong m_width= net_field_length((uchar **)&tmp);
  tmp+= (m_width + 7) / 8;

  if (type == UPDATE_ROWS_EVENT_V1 || type == UPDATE_ROWS_EVENT)
  {
    tmp+= (m_width + 7) / 8;
  }

  if (unlikely(end <= tmp))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  uint32 un_len= binlog_get_uncompress_len(tmp);
  if (unlikely(un_len == 0))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  if (unlikely(un_len > MAX_MAX_ALLOWED_PACKET))
    return ER_TOO_BIG_FOR_UNCOMPRESS;

  int32 comp_len= (int32)(len - (tmp - src) -
                          (contain_checksum ? BINLOG_CHECKSUM_LEN : 0));
  if (unlikely(comp_len <= 0))
    return ER_BINLOG_UNCOMPRESS_ERROR;

  *newlen= ulong(tmp - src) + un_len;
  if (contain_checksum)
    *newlen+= BINLOG_CHECKSUM_LEN;

  size_t alloc_size= ALIGN_SIZE(*newlen);
  
  *is_malloc= false;
  if (alloc_size <= buf_size) 
  {
    new_dst= buf;
  }
  else
  {
    new_dst= (uchar*) my_malloc(PSI_INSTRUMENT_ME, alloc_size, MYF(MY_WME));
    if (!new_dst)
      return ER_BINLOG_UNCOMPRESS_ERROR;
    *is_malloc= true;
  }

  /* Copy the head. */
  memcpy(new_dst, src , tmp - src);
  /* Uncompress the body. */
  if (binlog_buf_uncompress(tmp, new_dst + (tmp - src),
                            comp_len, &un_len))
  {
    if (*is_malloc)
      my_free(new_dst);
    return ER_BINLOG_UNCOMPRESS_ERROR;
  }

  new_dst[EVENT_TYPE_OFFSET]= type;
  int4store(new_dst + EVENT_LEN_OFFSET, *newlen);
  if (contain_checksum)
  {
    ulong clear_len= *newlen - BINLOG_CHECKSUM_LEN;
    int4store(new_dst + clear_len,
              my_checksum(0L, (uchar *)new_dst, clear_len));
  }
  *dst= new_dst;
  return 0;
}

/**
  Get the length of uncompress content.
  return 0 means error.
*/

uint32 binlog_get_uncompress_len(const uchar *buf)
{
  uint32 len, lenlen;

  if ((buf == NULL) || ((buf[0] & 0xe0) != 0x80))
    return 0;

  lenlen= buf[0] & 0x07;

  buf++;
  /* Length is stored in high byte first order, like myisam keys */
  switch(lenlen) {
  case 1:
    len= buf[0];
    break;
  case 2:
    len= mi_uint2korr(buf);
    break;
  case 3:
    len= mi_uint3korr(buf);
    break;
  case 4:
    len= mi_uint4korr(buf);
    break;
  default:
    DBUG_ASSERT(lenlen >= 1 && lenlen <= 4);
    len= 0;
    break;
  }
  return len;
}

/**
   Uncompress the content in 'src' with length of 'len' to 'dst'.

   Note: 1) Then the caller should guarantee the length of 'dst' (which
      can be got by statement_get_uncompress_len) is enough to hold
      the content uncompressed.
         2) The 'newlen' should stored the length of 'dst', and it will
      be set as the size of uncompressed content after return.

   return zero if successful, others otherwise.
*/
int binlog_buf_uncompress(const uchar *src, uchar *dst, uint32 len,
                          uint32 *newlen)
{
  if ((src[0] & 0x80) == 0)
    return 1;

  uint32 lenlen= src[0] & 0x07;
  uLongf buflen= *newlen;                       // zlib type

  uint32 alg= (src[0] & 0x70) >> 4;
  switch(alg) {
  case 0:
    // zlib
    if (uncompress((Bytef *)dst, &buflen,
      (const Bytef*)src + 1 + lenlen, len - 1 - lenlen) != Z_OK)
      return 1;
    break;
  default:
    //TODO
    //bad algorithm
    return 1;
  }

  DBUG_ASSERT(*newlen == (uint32)buflen);
  *newlen= (uint32)buflen;
  return 0;
}


/**************************************************************************
 Log_event methods (= the parent class of all events)
**************************************************************************/

/**
  @return
  returns the human readable name of the event's type
*/

const char* Log_event::get_type_str(Log_event_type type)
{
  switch(type) {
  case START_EVENT_V3:  return "Start_v3";
  case STOP_EVENT:   return "Stop";
  case QUERY_EVENT:  return "Query";
  case ROTATE_EVENT: return "Rotate";
  case INTVAR_EVENT: return "Intvar";
  case LOAD_EVENT:   return "Load";
  case NEW_LOAD_EVENT:   return "New_load";
  case SLAVE_EVENT:  return "Slave";
  case CREATE_FILE_EVENT: return "Create_file";
  case APPEND_BLOCK_EVENT: return "Append_block";
  case DELETE_FILE_EVENT: return "Delete_file";
  case EXEC_LOAD_EVENT: return "Exec_load";
  case RAND_EVENT: return "RAND";
  case XID_EVENT: return "Xid";
  case USER_VAR_EVENT: return "User var";
  case FORMAT_DESCRIPTION_EVENT: return "Format_desc";
  case TABLE_MAP_EVENT: return "Table_map";
  case PRE_GA_WRITE_ROWS_EVENT: return "Write_rows_event_old";
  case PRE_GA_UPDATE_ROWS_EVENT: return "Update_rows_event_old";
  case PRE_GA_DELETE_ROWS_EVENT: return "Delete_rows_event_old";
  case WRITE_ROWS_EVENT_V1: return "Write_rows_v1";
  case UPDATE_ROWS_EVENT_V1: return "Update_rows_v1";
  case DELETE_ROWS_EVENT_V1: return "Delete_rows_v1";
  case WRITE_ROWS_EVENT: return "Write_rows";
  case UPDATE_ROWS_EVENT: return "Update_rows";
  case DELETE_ROWS_EVENT: return "Delete_rows";
  case BEGIN_LOAD_QUERY_EVENT: return "Begin_load_query";
  case EXECUTE_LOAD_QUERY_EVENT: return "Execute_load_query";
  case INCIDENT_EVENT: return "Incident";
  case ANNOTATE_ROWS_EVENT: return "Annotate_rows";
  case BINLOG_CHECKPOINT_EVENT: return "Binlog_checkpoint";
  case GTID_EVENT: return "Gtid";
  case GTID_LIST_EVENT: return "Gtid_list";
  case START_ENCRYPTION_EVENT: return "Start_encryption";

  /* The following is only for mysqlbinlog */
  case IGNORABLE_LOG_EVENT: return "Ignorable log event";
  case ROWS_QUERY_LOG_EVENT: return "MySQL Rows_query";
  case GTID_LOG_EVENT: return "MySQL Gtid";
  case ANONYMOUS_GTID_LOG_EVENT: return "MySQL Anonymous_Gtid";
  case PREVIOUS_GTIDS_LOG_EVENT: return "MySQL Previous_gtids";
  case HEARTBEAT_LOG_EVENT: return "Heartbeat";
  case TRANSACTION_CONTEXT_EVENT: return "Transaction_context";
  case VIEW_CHANGE_EVENT: return "View_change";
  case XA_PREPARE_LOG_EVENT: return "XA_prepare";
  case PARTIAL_UPDATE_ROWS_EVENT: return "MySQL Update_rows_partial";
  case TRANSACTION_PAYLOAD_EVENT: return "MySQL Transaction_payload";
  case HEARTBEAT_LOG_EVENT_V2: return "MySQL Heartbeat";
  case QUERY_COMPRESSED_EVENT: return "Query_compressed";
  case WRITE_ROWS_COMPRESSED_EVENT: return "Write_rows_compressed";
  case UPDATE_ROWS_COMPRESSED_EVENT: return "Update_rows_compressed";
  case DELETE_ROWS_COMPRESSED_EVENT: return "Delete_rows_compressed";
  case WRITE_ROWS_COMPRESSED_EVENT_V1: return "Write_rows_compressed_v1";
  case UPDATE_ROWS_COMPRESSED_EVENT_V1: return "Update_rows_compressed_v1";
  case DELETE_ROWS_COMPRESSED_EVENT_V1: return "Delete_rows_compressed_v1";
  case PARTIAL_ROW_DATA_EVENT: return "Partial_rows";

  default: return "Unknown";    /* impossible */
  }
}

const char* Log_event::get_type_str()
{
  return get_type_str(get_type_code());
}


/*
  Log_event::Log_event()
*/

Log_event::Log_event(const uchar *buf,
                     const Format_description_log_event* description_event)
  :temp_buf(0), exec_time(0), cache_type(Log_event::EVENT_INVALID_CACHE)
#ifndef MYSQL_CLIENT
    , slave_exec_mode(SLAVE_EXEC_MODE_STRICT)
#endif
{
#ifndef MYSQL_CLIENT
  thd= 0;
#endif
  when= uint4korr(buf);
  when_sec_part= ~0UL;
  server_id= uint4korr(buf + SERVER_ID_OFFSET);
  data_written= uint4korr(buf + EVENT_LEN_OFFSET);
  log_pos= uint4korr(buf + LOG_POS_OFFSET);
  DBUG_PRINT("info", ("log_pos: %llu", log_pos));

  flags= uint2korr(buf + FLAGS_OFFSET);
  if (((uchar)buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT) ||
      ((uchar)buf[EVENT_TYPE_OFFSET] == ROTATE_EVENT))
  {
    /*
      These events always have a header which stops here (i.e. their
      header is FROZEN).
    */
    /*
      Initialization to zero of all other Log_event members as they're
      not specified. Currently there are no such members; in the future
      there will be an event UID (but Format_description and Rotate
      don't need this UID, as they are not propagated through
      --log-slave-updates (remember the UID is used to not play a query
      twice when you have two masters which are slaves of a 3rd master).
      Then we are done.
    */
    return;
  }
  /* otherwise, go on with reading the header from buf (nothing now) */
}


/**
  This needn't be format-tolerant, because we only parse the first
  LOG_EVENT_MINIMAL_HEADER_LEN bytes (just need the event's length).
*/

int Log_event::read_log_event(IO_CACHE* file, String* packet,
                              const Format_description_log_event *fdle,
                              enum_binlog_checksum_alg checksum_alg_arg,
                              size_t max_allowed_packet)
{
  ulong data_len;
  char buf[LOG_EVENT_MINIMAL_HEADER_LEN];
  uchar ev_offset= packet->length();

  DBUG_ENTER("Log_event::read_log_event(IO_CACHE*,String*...)");

  if (my_b_read(file, (uchar*) buf, sizeof(buf)))
  {
    /*
      If the read hits eof, we must report it as eof so the caller
      will know it can go into cond_wait to be woken up on the next
      update to the log.
    */
    DBUG_PRINT("error",("file->error: %d", file->error));
    DBUG_RETURN(file->error == 0 ? LOG_READ_EOF :
                file->error > 0 ? LOG_READ_TRUNC : LOG_READ_IO);
  }
  data_len= uint4korr(buf + EVENT_LEN_OFFSET);

  /* Append the log event header to packet */
  if (packet->append(buf, sizeof(buf)))
    DBUG_RETURN(LOG_READ_MEM);

  if (data_len < LOG_EVENT_MINIMAL_HEADER_LEN)
    DBUG_RETURN(LOG_READ_BOGUS);

  if (data_len > MY_MAX(max_allowed_packet,
                        opt_binlog_rows_event_max_size + MAX_LOG_EVENT_HEADER))
    DBUG_RETURN(LOG_READ_TOO_LARGE);

  if (likely(data_len > LOG_EVENT_MINIMAL_HEADER_LEN))
  {
    /* Append rest of event, read directly from file into packet */
    if (packet->append(file, data_len - LOG_EVENT_MINIMAL_HEADER_LEN))
    {
      /*
        Fatal error occurred when appending rest of the event
        to packet, possible failures:
 1. EOF occurred when reading from file, it's really an error
           as there's supposed to be more bytes available.
           file->error will have been set to number of bytes left to read
        2. Read was interrupted, file->error would normally be set to -1
        3. Failed to allocate memory for packet, my_errno
           will be ENOMEM(file->error should be 0, but since the
           memory allocation occurs before the call to read it might
           be uninitialized)
      */
      DBUG_RETURN(my_errno == ENOMEM ? LOG_READ_MEM :
                  (file->error >= 0 ? LOG_READ_TRUNC: LOG_READ_IO));
    }
  }

  if (fdle->crypto_data.scheme)
  {
    uchar iv[BINLOG_IV_LENGTH];
    fdle->crypto_data.set_iv(iv, (uint32) (my_b_tell(file) - data_len));
    size_t sz= data_len + ev_offset + 1;
#ifdef HAVE_WOLFSSL
    /*
      Workaround for MDEV-19582.
      WolfSSL reads memory out of bounds with decryption/NOPAD)
      We allocate a little more memory therefore.
    */
    sz+= MY_AES_BLOCK_SIZE;
#endif
    char *newpkt= (char*)my_malloc(PSI_INSTRUMENT_ME, sz, MYF(MY_WME));
    if (!newpkt)
      DBUG_RETURN(LOG_READ_MEM);
    memcpy(newpkt, packet->ptr(), ev_offset);

    uint dstlen= (uint) sz - ev_offset - 4;
    uchar *src= (uchar*)packet->ptr() + ev_offset;
    uchar *dst= (uchar*)newpkt + ev_offset;
    memcpy(src + EVENT_LEN_OFFSET, src, 4);
    if (encryption_crypt(src + 4, data_len - 4, dst + 4, &dstlen,
            fdle->crypto_data.key, fdle->crypto_data.key_length, iv,
            sizeof(iv), ENCRYPTION_FLAG_DECRYPT | ENCRYPTION_FLAG_NOPAD,
            ENCRYPTION_KEY_SYSTEM_DATA, fdle->crypto_data.key_version))
    {
      my_free(newpkt);
      DBUG_RETURN(LOG_READ_DECRYPT);
    }
    DBUG_ASSERT(dstlen == data_len - 4);
    memcpy(dst, dst + EVENT_LEN_OFFSET, 4);
    int4store(dst + EVENT_LEN_OFFSET, data_len);
    packet->reset(newpkt, data_len + ev_offset, data_len + ev_offset + 1,
                  &my_charset_bin);
  }

  /*
    CRC verification of the Dump thread
  */
  if (data_len > LOG_EVENT_MINIMAL_HEADER_LEN)
  {
    /* Corrupt the event for Dump thread*/
    DBUG_EXECUTE_IF("corrupt_read_log_event2",
      uchar *debug_event_buf_c= (uchar*) packet->ptr() + ev_offset;
      if (debug_event_buf_c[EVENT_TYPE_OFFSET] != FORMAT_DESCRIPTION_EVENT)
      {
        int debug_cor_pos= rand() % (data_len - BINLOG_CHECKSUM_LEN);
        debug_event_buf_c[debug_cor_pos] =~ debug_event_buf_c[debug_cor_pos];
        DBUG_PRINT("info", ("Corrupt the event at Log_event::read_log_event: byte on position %d", debug_cor_pos));
        DBUG_SET("-d,corrupt_read_log_event2");
      }
    );
    if (event_checksum_test((uchar*) packet->ptr() + ev_offset,
                             data_len, checksum_alg_arg))
      DBUG_RETURN(LOG_READ_CHECKSUM_FAILURE);
  }
  DBUG_RETURN(0);
}

Log_event* Log_event::read_log_event(IO_CACHE* file, int *out_error,
                                     const Format_description_log_event *fdle,
                                     my_bool crc_check, my_bool print_errors,
                                     size_t max_allowed_packet)
{
  DBUG_ENTER("Log_event::read_log_event(IO_CACHE*,Format_description_log_event*...)");
  DBUG_ASSERT(fdle != 0);
  String event;
  const char *error= 0;
  Log_event *res= 0;

  *out_error= 0;
  switch (read_log_event(file, &event, fdle, BINLOG_CHECKSUM_ALG_OFF,
                         max_allowed_packet))
  {
    case 0:
      break;
    case LOG_READ_EOF: // no error here; we are at the file's end
      goto err;
    case LOG_READ_BOGUS:
      error= "Event invalid";
      goto err;
    case LOG_READ_IO:
      error= "read error";
      goto err;
    case LOG_READ_MEM:
      error= "Out of memory";
      goto err;
    case LOG_READ_TRUNC:
      error= "Event truncated";
      goto err;
    case LOG_READ_TOO_LARGE:
      error= "Event too big";
      goto err;
    case LOG_READ_DECRYPT:
      error= "Event decryption failure";
      goto err;
    case LOG_READ_CHECKSUM_FAILURE:
    default:
      DBUG_ASSERT(0);
      error= "internal error";
      goto err;
  }

  /*
    print_errors is false to prevent redundant error messages cluttering up the
    log, as it will be printed below (if _our_ print_errors is true)
  */
  if ((res= read_log_event((uchar*) event.ptr(), event.length(),
                           &error, fdle, crc_check, false)))
    res->register_temp_buf((uchar*) event.release(), true);

err:
  if (unlikely(error))
  {
    DBUG_ASSERT(!res);
#ifdef MYSQL_CLIENT
    if (force_opt)
      DBUG_RETURN(new Unknown_log_event());
#endif

    /*
      The SQL slave thread will check *out_error to know
      if there was an I/O error. Even if there is no "low-level" I/O errors
      with 'file', any of the high-level above errors is worrying
      enough to stop the SQL thread now ; as we are skipping the current event,
      going on with reading and successfully executing other events can
      only corrupt the slave's databases. So stop.
    */
    *out_error= 1;
    /*
      Clear any error that might have been set in the IO_CACHE from a read
      error, while we are still holding the relay log mutex (if reading from
      the hot log). Otherwise the error might interfere unpredictably with
      write operations to the same IO_CACHE in the IO thread.
    */
    file->error= 0;


#ifndef MYSQL_CLIENT
    if (!print_errors)
      DBUG_RETURN(res);
#endif

    if (event.length() >= LOG_EVENT_MINIMAL_HEADER_LEN)
      sql_print_error("Error in Log_event::read_log_event(): '%s',"
                      " data_len: %lu, event_type: %u", error,
                      (ulong) uint4korr(&event[EVENT_LEN_OFFSET]),
                      (uint) (uchar)event[EVENT_TYPE_OFFSET]);
    else
      sql_print_error("Error in Log_event::read_log_event(): '%s'", error);
  }
  DBUG_RETURN(res);
}

/**
  Binlog format tolerance is in (buf, event_len, fdle)
  constructors.
*/

Log_event* Log_event::read_log_event(const uchar *buf, size_t event_len,
                                     const char **error,
                                     const Format_description_log_event *fdle,
                                     my_bool crc_check,
                                     my_bool print_errors)
{
  Log_event* ev= nullptr;
  enum_binlog_checksum_alg alg;
  DBUG_ENTER("Log_event::read_log_event(char*,...)");
  DBUG_ASSERT(fdle != 0);
  DBUG_PRINT("info", ("binlog_version: %d", fdle->binlog_version));
  DBUG_DUMP_EVENT_BUF(buf, event_len);

  *error= 0;
  /*
    Check the integrity; This is needed because handle_slave_io() doesn't
    check if packet is of proper length.
 */
  if (event_len < EVENT_LEN_OFFSET)
  {
    *error="Sanity check failed";  // Needed to free buffer
    DBUG_RETURN(NULL); // general sanity check - will fail on a partial read
  }

  uint event_type= buf[EVENT_TYPE_OFFSET];
  switch (event_type) {
  case FORMAT_DESCRIPTION_EVENT:
    // If event is FD the descriptor is in it.
    if (unlikely(get_checksum_alg(buf, event_len, &alg)))
    {
#ifdef MYSQL_CLIENT
      if (force_opt)
      {
        event_len-= BINLOG_CHECKSUM_LEN;
        ev= new Unknown_log_event(buf, fdle);
        goto exit;
      }
#endif
      *error= "Found invalid event in binary log";
      DBUG_RETURN(nullptr);
    }
    break;
  case START_EVENT_V3:
  // all following START events in the current file are without checksum
    (const_cast< Format_description_log_event *>(fdle))->used_checksum_alg=
      BINLOG_CHECKSUM_ALG_OFF;
    // fall-through
  default:
  /*
    CRC verification by SQL and Show-Binlog-Events master side.
    The caller has to provide @fdle->checksum_alg to
    be the last seen FD's (A) descriptor.
    Notice, FD of the binlog can be only in one instance and therefore
    Show-Binlog-Events executing master side thread needs just to know
    the only FD's (A) value -  whereas RL can contain more.
    In the RL case, the alg is kept in FD_e (@fdle) which is reset
    to the newer read-out event after its execution with possibly new alg descriptor.
    Therefore in a typical sequence of RL:
    {FD_s^0, FD_m, E_m^1} E_m^1 
    will be verified with (A) of FD_m.

    See legends definition on MYSQL_BIN_LOG::relay_log_checksum_alg docs
    lines (log.h).

    Notice, a pre-checksum FD version forces alg := BINLOG_CHECKSUM_ALG_UNDEF.
  */
    alg= fdle->used_checksum_alg;
  // Emulate the corruption during reading an event
  DBUG_EXECUTE_IF("corrupt_read_log_event_char",
    {
      uchar *debug_event_buf_c= const_cast<uchar*>(buf);
      int debug_cor_pos= rand() % (event_len - BINLOG_CHECKSUM_LEN);
      debug_event_buf_c[debug_cor_pos]=~ debug_event_buf_c[debug_cor_pos];
      DBUG_PRINT("info", ("Corrupt the event at Log_event::read_log_event(char*,...): byte on position %d", debug_cor_pos));
      DBUG_SET("-d,corrupt_read_log_event_char");
    }
  );                                                 
  }
  if (crc_check && event_checksum_test(const_cast<uchar*>(buf), event_len, alg))
  {
#ifdef MYSQL_CLIENT
    if (force_opt)
    {
      event_len-= BINLOG_CHECKSUM_LEN;
      ev= new Unknown_log_event(buf, fdle);
      goto exit;
    }
    *error= "Event crc check failed! Most likely there is event corruption.";
#else
    *error= ER_THD_OR_DEFAULT(current_thd, ER_BINLOG_READ_EVENT_CHECKSUM_FAILURE);
    if (print_errors)
      sql_print_error("%s", *error);
#endif
    DBUG_RETURN(NULL);
  }

  if (event_type > fdle->number_of_event_types &&
      event_type != FORMAT_DESCRIPTION_EVENT)
  {
    /*
      It is unsafe to use the fdle if its post_header_len
      array does not include the event type.
    */
    DBUG_PRINT("error", ("event type %d found, but the current "
                         "Format_description_log_event supports only %d event "
                         "types", event_type,
                         fdle->number_of_event_types));
    ev= NULL;
  }
  else
  {
    /*
      In some previuos versions (see comment in
      Format_description_log_event::Format_description_log_event(char*,...)),
      event types were assigned different id numbers than in the
      present version. In order to replicate from such versions to the
      present version, we must map those event type id's to our event
      type id's.  The mapping is done with the event_type_permutation
      array, which was set up when the Format_description_log_event
      was read.
    */
    if (fdle->event_type_permutation)
    {
      int new_event_type= fdle->event_type_permutation[event_type];
      DBUG_PRINT("info", ("converting event type %d to %d (%s)",
                   event_type, new_event_type,
                   get_type_str((Log_event_type)new_event_type)));
      event_type= new_event_type;
    }

    if (alg != BINLOG_CHECKSUM_ALG_UNDEF &&
        (event_type == FORMAT_DESCRIPTION_EVENT ||
         alg != BINLOG_CHECKSUM_ALG_OFF))
      event_len= event_len - BINLOG_CHECKSUM_LEN;

    ev= Log_event::read_log_event_no_checksum(buf, event_len, error, fdle);
  }

#ifdef MYSQL_CLIENT
exit:
  if (ev)
  {
    ev->read_checksum_alg= alg;
    if (alg != BINLOG_CHECKSUM_ALG_OFF && alg != BINLOG_CHECKSUM_ALG_UNDEF)
      ev->read_checksum_value= uint4korr(buf + (event_len));
  }
#endif

  DBUG_RETURN(ev);
}

Log_event *Log_event::read_log_event_no_checksum(
    const uchar *buf, size_t event_len, const char **error,
    const Format_description_log_event *fdle)
{
  Log_event* ev;
  DBUG_ENTER("Log_event::read_log_event_no_checksum(char*,...)");
  DBUG_ASSERT(fdle != 0);
  DBUG_PRINT("info", ("binlog_version: %d", fdle->binlog_version));
  DBUG_DUMP_EVENT_BUF(buf, event_len);

  *error= 0;
  /*
    Check the integrity; This is needed because handle_slave_io() doesn't
    check if packet is of proper length.
 */
  if (event_len < EVENT_LEN_OFFSET)
  {
    *error="Sanity check failed";  // Needed to free buffer
    DBUG_RETURN(NULL); // general sanity check - will fail on a partial read
  }

  uint event_type= buf[EVENT_TYPE_OFFSET];

  /*
    In some previuos versions (see comment in
    Format_description_log_event::Format_description_log_event(char*,...)),
    event types were assigned different id numbers than in the
    present version. In order to replicate from such versions to the
    present version, we must map those event type id's to our event
    type id's.  The mapping is done with the event_type_permutation
    array, which was set up when the Format_description_log_event
    was read.
  */
  if (fdle->event_type_permutation)
  {
    int new_event_type= fdle->event_type_permutation[event_type];
    DBUG_PRINT("info", ("converting event type %d to %d (%s)",
                 event_type, new_event_type,
                 get_type_str((Log_event_type)new_event_type)));
    event_type= new_event_type;
  }


    /*
      Create an object of Ignorable_log_event for unrecognized sub-class.
      So that SLAVE SQL THREAD will only update the position and continue.
      We should look for this flag first instead of judging by event_type
      Any event can be Ignorable_log_event if it has this flag on.
      look into @note of Ignorable_log_event
    */
    if (uint2korr(buf + FLAGS_OFFSET) & LOG_EVENT_IGNORABLE_F)
    {
      ev= new Ignorable_log_event(buf, fdle,
                                  get_type_str((Log_event_type) event_type));
      goto exit;
    }
    switch(event_type) {
    case QUERY_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Query_log_event(buf, static_cast<uint>(event_len), fdle,
                              QUERY_EVENT);
      break;
    case QUERY_COMPRESSED_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Query_compressed_log_event(buf, static_cast<uint>(event_len),
                                         fdle, QUERY_COMPRESSED_EVENT);
      break;
    case ROTATE_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Rotate_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case BINLOG_CHECKPOINT_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Binlog_checkpoint_log_event(buf, static_cast<uint>(event_len),
                                          fdle);
      break;
    case GTID_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Gtid_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case GTID_LIST_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Gtid_list_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case APPEND_BLOCK_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Append_block_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case DELETE_FILE_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Delete_file_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case STOP_EVENT:
      ev= new Stop_log_event(buf, fdle);
      break;
    case INTVAR_EVENT:
      ev= new Intvar_log_event(buf, fdle);
      break;
    case XID_EVENT:
      ev= new Xid_log_event(buf, fdle);
      break;
    case XA_PREPARE_LOG_EVENT:
      ev= new XA_prepare_log_event(buf, fdle);
      break;
    case RAND_EVENT:
      ev= new Rand_log_event(buf, fdle);
      break;
    case USER_VAR_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new User_var_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case FORMAT_DESCRIPTION_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Format_description_log_event(buf, static_cast<uint>(event_len),
                                           fdle);
      break;
#if defined(HAVE_REPLICATION) 
    case WRITE_ROWS_EVENT_V1:
    case WRITE_ROWS_EVENT:
      ev= new Write_rows_log_event(buf, event_len, fdle);
      break;
    case UPDATE_ROWS_EVENT_V1:
    case UPDATE_ROWS_EVENT:
      ev= new Update_rows_log_event(buf, event_len, fdle);
      break;
    case DELETE_ROWS_EVENT_V1:
    case DELETE_ROWS_EVENT:
      ev= new Delete_rows_log_event(buf, event_len, fdle);
      break;

    case WRITE_ROWS_COMPRESSED_EVENT:
    case WRITE_ROWS_COMPRESSED_EVENT_V1:
      ev= new Write_rows_compressed_log_event(buf, event_len, fdle);
      break;
    case UPDATE_ROWS_COMPRESSED_EVENT:
    case UPDATE_ROWS_COMPRESSED_EVENT_V1:
      ev= new Update_rows_compressed_log_event(buf, event_len, fdle);
      break;
    case DELETE_ROWS_COMPRESSED_EVENT:
    case DELETE_ROWS_COMPRESSED_EVENT_V1:
      ev= new Delete_rows_compressed_log_event(buf, event_len, fdle);
      break;

      /* MySQL GTID events are ignored */
    case GTID_LOG_EVENT:
    case ANONYMOUS_GTID_LOG_EVENT:
    case PREVIOUS_GTIDS_LOG_EVENT:
    case TRANSACTION_CONTEXT_EVENT:
    case HEARTBEAT_LOG_EVENT_V2:                // MySQL 8.0
    case VIEW_CHANGE_EVENT:
      ev= new Ignorable_log_event(buf, fdle,
                                  get_type_str((Log_event_type) event_type));
      break;

    case TABLE_MAP_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Table_map_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
#endif
  case PARTIAL_ROW_DATA_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Partial_rows_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case BEGIN_LOAD_QUERY_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Begin_load_query_log_event(buf, static_cast<uint>(event_len),
                                         fdle);
      break;
    case EXECUTE_LOAD_QUERY_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Execute_load_query_log_event(buf, static_cast<uint>(event_len),
                                           fdle);
      break;
    case INCIDENT_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Incident_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case ANNOTATE_ROWS_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Annotate_rows_log_event(buf, static_cast<uint>(event_len), fdle);
      break;
    case START_ENCRYPTION_EVENT:
      DBUG_ASSERT(event_len <= UINT32_MAX);
      ev= new Start_encryption_log_event(buf, static_cast<uint>(event_len),
                                         fdle);
      break;
    case TRANSACTION_PAYLOAD_EVENT:             // MySQL 8.0
      *error=
        "Found incompatible MySQL 8.0 TRANSACTION_PAYLOAD_EVENT event. "
        "You can avoid this event by specifying "
        "'binlog_transaction_compression=0' in the MySQL server";
      ev= NULL;
      break;
    case PARTIAL_UPDATE_ROWS_EVENT:             // MySQL 8.0
      *error=
        "Found incompatible MySQL 8.0 PARTIAL_UPDATE_ROWS_EVENT event. "
        "You can avoid this event by specifying "
        "'binlog-row-value-options=\"\"' in the MySQL server";
      ev= NULL;
      break;

    case PRE_GA_WRITE_ROWS_EVENT:
    case PRE_GA_UPDATE_ROWS_EVENT:
    case PRE_GA_DELETE_ROWS_EVENT:
    case START_EVENT_V3: /* this is sent only by MySQL <=4.x */
    case CREATE_FILE_EVENT:
    case EXEC_LOAD_EVENT:
    case LOAD_EVENT:
    case NEW_LOAD_EVENT:
    default:
      DBUG_PRINT("error",("Unknown event code: %d",
                          (uchar) buf[EVENT_TYPE_OFFSET]));
      ev= NULL;
      break;
  }
exit:

  DBUG_PRINT("read_event", ("%s(type_code: %u; event_len: %zu)",
                            ev ? ev->get_type_str() : "<unknown>",
                            (uchar)buf[EVENT_TYPE_OFFSET],
                            event_len));
  /*
    is_valid() are small event-specific sanity tests which are
    important; for example there are some my_malloc() in constructors
    (e.g. Query_log_event::Query_log_event(char*...)); when these
    my_malloc() fail we can't return an error out of the constructor
    (because constructor is "void") ; so instead we leave the pointer we
    wanted to allocate (e.g. 'query') to 0 and we test it in is_valid().
    Same for Format_description_log_event, member 'post_header_len'.

    SLAVE_EVENT is never used, so it should not be read ever.
  */
  if (!ev || !ev->is_valid() || (event_type == SLAVE_EVENT))
  {
    DBUG_PRINT("error",("Found invalid event in binary log"));

    delete ev;
#ifdef MYSQL_CLIENT
    if (!force_opt) /* then mysqlbinlog dies */
    {
      if (!*error)
        *error= "Found invalid event in binary log";
      DBUG_RETURN(0);
    }
    ev= new Unknown_log_event(buf, fdle);
#else
    if (!*error)
      *error= "Found invalid event in binary log";
    DBUG_RETURN(0);
#endif
  }
  DBUG_RETURN(ev);  
}


int handler_binlog_reader::read_log_event(String *packet, uint32_t ev_offset,
                                          size_t max_allowed)
{
  uint32_t sofar= 0;
  bool header_read= false;
  uint32_t target_size= EVENT_LEN_OFFSET + 4;
  int res;

  if (unlikely(!buf))
    return LOG_READ_MEM;

  /*
    Loop, first reading the "length" field, and then continuing to read data
    until a full event has been placed in the packet.
  */
  for (;;)
  {
    if (buf_data_remain <= 0)
    {
      res= read_binlog_data(buf, BUF_SIZE);
      if (res <= 0)
      {
        res= (res < 0 ? LOG_READ_IO : LOG_READ_EOF);
        goto err;
      }
      buf_data_pos= 0;
      buf_data_remain= res;
    }
    uint32_t amount= std::min(target_size - sofar, buf_data_remain);
    packet->append((char *)buf + buf_data_pos, amount);
    buf_data_pos+= amount;
    buf_data_remain-= amount;
    sofar+= amount;
    if (target_size == sofar)
    {
      if (header_read)
        break;
      else
      {
        header_read= true;
        target_size= uint4korr(&((*packet)[EVENT_LEN_OFFSET + ev_offset]));
        if (target_size < LOG_EVENT_MINIMAL_HEADER_LEN)
        {
          res= LOG_READ_BOGUS;
          goto err;
        }
 else if (target_size > max_allowed)
        {
          res= LOG_READ_TOO_LARGE;
          goto err;
        }
 /*
          Note that here we rely on the fact that all valid events have more
          data after the length. This way we avoid conditional for the
          (useless) special case where we don't need to read anything more
          after having read the first part.
        */
        DBUG_ASSERT(LOG_EVENT_MINIMAL_HEADER_LEN > EVENT_LEN_OFFSET+4);
      }
    }
  }
  res= 0;  /* Success */
err:
  return res;
}


/* 2 utility functions for the next method */

/**
   Read a string with length from memory.

   This function reads the string-with-length stored at
   <code>src</code> and extract the length into <code>*len</code> and
   a pointer to the start of the string into <code>*dst</code>. The
   string can then be copied using <code>memcpy()</code> with the
   number of bytes given in <code>*len</code>.

   @param src Pointer to variable holding a pointer to the memory to
              read the string from.
   @param dst Pointer to variable holding a pointer where the actual
              string starts. Starting from this position, the string
              can be copied using @c memcpy().
   @param len Pointer to variable where the length will be stored.
   @param end One-past-the-end of the memory where the string is
              stored.

   @return    Zero if the entire string can be copied successfully,
              @c UINT_MAX if the length could not be read from memory
              (that is, if <code>*src >= end</code>), otherwise the
              number of bytes that are missing to read the full
              string, which happends <code>*dst + *len >= end</code>.
*/
static int
get_str_len_and_pointer(const Log_event::Byte **src,
                        const char **dst,
                        uint *len,
                        const Log_event::Byte *end)
{
  if (*src >= end)
    return -1;       // Will be UINT_MAX in two-complement arithmetics
  uint length= **src;
  if (length > 0)
  {
    if (*src + length >= end)
      return (int)(*src + length - end + 1);   // Number of bytes missing
    *dst= (char *)*src + 1;                    // Will be copied later
  }
  *len= length;
  *src+= length + 1;
  return 0;
}

static void copy_str_and_move(const char **src, Log_event::Byte **dst, 
                              size_t len)
{
  memcpy(*dst, *src, len);
  *src= (const char *)*dst;
  (*dst)+= len;
  *(*dst)++= 0;
}


#ifdef DBUG_TRACE
static char const *
code_name(int code)
{
  static char buf[255];
  switch (code) {
  case Q_FLAGS2_CODE: return "Q_FLAGS2_CODE";
  case Q_SQL_MODE_CODE: return "Q_SQL_MODE_CODE";
  case Q_CATALOG_CODE: return "Q_CATALOG_CODE";
  case Q_AUTO_INCREMENT: return "Q_AUTO_INCREMENT";
  case Q_CHARSET_CODE: return "Q_CHARSET_CODE";
  case Q_TIME_ZONE_CODE: return "Q_TIME_ZONE_CODE";
  case Q_CATALOG_NZ_CODE: return "Q_CATALOG_NZ_CODE";
  case Q_LC_TIME_NAMES_CODE: return "Q_LC_TIME_NAMES_CODE";
  case Q_CHARSET_DATABASE_CODE: return "Q_CHARSET_DATABASE_CODE";
  case Q_TABLE_MAP_FOR_UPDATE_CODE: return "Q_TABLE_MAP_FOR_UPDATE_CODE";
  case Q_MASTER_DATA_WRITTEN_CODE: return "Q_MASTER_DATA_WRITTEN_CODE";
  case Q_HRNOW: return "Q_HRNOW";
  case Q_XID: return "XID";
  case Q_GTID_FLAGS3: return "Q_GTID_FLAGS3";
  case Q_CHARACTER_SET_COLLATIONS: return "Q_CHARACTER_SET_COLLATIONS";
  }
  snprintf(buf, sizeof(buf), "CODE#%d", code);
  return buf;
}
#endif

#define VALIDATE_BYTES_READ(CUR_POS, START, EVENT_LEN)      \
  do {                                                      \
       uchar *cur_pos= (uchar *)CUR_POS;                    \
       uchar *start= (uchar *)START;                        \
       uint len= EVENT_LEN;                                 \
       uint bytes_read= (uint)(cur_pos - start);            \
       DBUG_PRINT("info", ("Bytes read: %u event_len:%u.\n",\
             bytes_read, len));                             \
       if (bytes_read >= len)                               \
         DBUG_VOID_RETURN;                                  \
  } while (0)

/**
   Macro to check that there is enough space to read from memory.

   @param PTR Pointer to memory
   @param END End of memory
   @param CNT Number of bytes that should be read.
 */
#define CHECK_SPACE(PTR,END,CNT)                      \
  do {                                                \
    DBUG_PRINT("info", ("Read %s", code_name(pos[-1]))); \
    if ((PTR) + (CNT) > (END)) {                      \
      DBUG_PRINT("info", ("query= 0"));               \
      query= 0;                                       \
      DBUG_VOID_RETURN;                               \
    }                                                 \
  } while (0)


/**
  This is used by the SQL slave thread to prepare the event before execution.
*/
Query_log_event::Query_log_event(const uchar *buf, uint event_len,
                                 const Format_description_log_event
                                 *description_event,
                                 Log_event_type event_type)
  :Log_event(buf, description_event), data_buf(0), query(NullS),
   db(NullS), catalog_len(0), status_vars_len(0),
   flags2_inited(0), sql_mode_inited(0), charset_inited(0),
   character_set_collations({0,0}), flags2(0),
   auto_increment_increment(1), auto_increment_offset(1),
   time_zone_len(0), lc_time_names_number(0), charset_database_number(0),
   table_map_for_update(0), xid(0), gtid_flags_extra(0),
   sa_seq_no(0)
{
  ulong data_len;
  uint8 common_header_len, post_header_len;
  Log_event::Byte *start;
  const Log_event::Byte *end;
  bool catalog_nz= 1;
  DBUG_ENTER("Query_log_event::Query_log_event(char*,...)");

  memset(&user, 0, sizeof(user));
  memset(&host, 0, sizeof(host));
  common_header_len= description_event->common_header_len;
  post_header_len= description_event->post_header_len[event_type-1];
  DBUG_PRINT("info",("event_len: %u  common_header_len: %d  post_header_len: %d",
                     event_len, common_header_len, post_header_len));

  /*
    We test if the event's length is sensible, and if so we compute data_len.
    We cannot rely on QUERY_HEADER_LEN here as it would not be format-tolerant.
    We use QUERY_HEADER_MINIMAL_LEN which is the same for 3.23, 4.0 & 5.0.
  */
  if (event_len < (uint)(common_header_len + post_header_len))
    DBUG_VOID_RETURN;
  data_len= event_len - (common_header_len + post_header_len);
  buf+= common_header_len;

  thread_id = slave_proxy_id = uint4korr(buf + Q_THREAD_ID_OFFSET);
  exec_time = uint4korr(buf + Q_EXEC_TIME_OFFSET);
  db_len = (uchar)buf[Q_DB_LEN_OFFSET]; // TODO: add a check of all *_len vars
  error_code = uint2korr(buf + Q_ERR_CODE_OFFSET);

  status_vars_len= uint2korr(buf + Q_STATUS_VARS_LEN_OFFSET);
  /*
    Check if status variable length is corrupt and will lead to very
    wrong data. We could be even more strict and require data_len to
    be even bigger, but this will suffice to catch most corruption
    errors that can lead to a crash.
  */
  if (status_vars_len > MY_MIN(data_len, MAX_SIZE_LOG_EVENT_STATUS))
  {
    DBUG_PRINT("info", ("status_vars_len (%u) > data_len (%lu); query= 0",
                        status_vars_len, data_len));
    query= 0;
    DBUG_VOID_RETURN;
  }
  data_len-= status_vars_len;
  DBUG_PRINT("info", ("Query_log_event has status_vars_len: %u",
                      (uint) status_vars_len));
  /*
    We have parsed everything we know in the post header for QUERY_EVENT,
    the rest of post header is either comes from older version MySQL or
    dedicated to derived events (e.g. Execute_load_query...)
  */

  /* variable-part: the status vars; only in MySQL 5.0  */
  
  start= (Log_event::Byte*) (buf+post_header_len);
  end= (const Log_event::Byte*) (start+status_vars_len);
  for (const Log_event::Byte* pos= start; pos < end;)
  {
    switch (*pos++) {
    case Q_FLAGS2_CODE:
      CHECK_SPACE(pos, end, 4);
      flags2_inited= description_event->options_written_to_bin_log;
      flags2= uint4korr(pos);
      DBUG_PRINT("info",("In Query_log_event, read flags2: %lu", (ulong) flags2));
      pos+= 4;
      break;
    case Q_SQL_MODE_CODE:
    {
      CHECK_SPACE(pos, end, 8);
      sql_mode_inited= 1;
      sql_mode= (sql_mode_t) uint8korr(pos);
      DBUG_PRINT("info",("In Query_log_event, read sql_mode: %llu", sql_mode));
      pos+= 8;
      break;
    }
    case Q_CATALOG_NZ_CODE:
      DBUG_PRINT("info", ("case Q_CATALOG_NZ_CODE; pos:%p; end:%p",
                          pos, end));
      if (get_str_len_and_pointer(&pos, &catalog, &catalog_len, end))
      {
        DBUG_PRINT("info", ("query= 0"));
        query= 0;
        DBUG_VOID_RETURN;
      }
      break;
    case Q_AUTO_INCREMENT:
      CHECK_SPACE(pos, end, 4);
      auto_increment_increment= uint2korr(pos);
      auto_increment_offset=    uint2korr(pos+2);
      pos+= 4;
      break;
    case Q_CHARSET_CODE:
    {
      CHECK_SPACE(pos, end, 6);
      charset_inited= 1;
      memcpy(charset, pos, 6);
      pos+= 6;
      break;
    }
    case Q_CHARACTER_SET_COLLATIONS:
    {
      const uchar *pos0= pos;
      CHECK_SPACE(pos, end, 1);
      uint16 count= *pos++;
      CHECK_SPACE(pos, end, count * 4);
      pos+= count * 4;
      character_set_collations= Lex_cstring((const char *) pos0,
                                            (const char *) pos);
      break;
    }
    case Q_TIME_ZONE_CODE:
    {
      if (get_str_len_and_pointer(&pos, &time_zone_str, &time_zone_len, end))
      {
        DBUG_PRINT("info", ("Q_TIME_ZONE_CODE: query= 0"));
        query= 0;
        DBUG_VOID_RETURN;
      }
      break;
    }
    case Q_CATALOG_CODE: /* for 5.0.x where 0<=x<=3 masters */
      CHECK_SPACE(pos, end, 1);
      if ((catalog_len= *pos))
        catalog= (char*) pos+1;                           // Will be copied later
      CHECK_SPACE(pos, end, catalog_len + 2);
      pos+= catalog_len+2; // leap over end 0
      catalog_nz= 0; // catalog has end 0 in event
      break;
    case Q_LC_TIME_NAMES_CODE:
      CHECK_SPACE(pos, end, 2);
      lc_time_names_number= uint2korr(pos);
      pos+= 2;
      break;
    case Q_CHARSET_DATABASE_CODE:
      CHECK_SPACE(pos, end, 2);
      charset_database_number= uint2korr(pos);
      pos+= 2;
      break;
    case Q_TABLE_MAP_FOR_UPDATE_CODE:
      CHECK_SPACE(pos, end, 8);
      table_map_for_update= uint8korr(pos);
      pos+= 8;
      break;
    case Q_MASTER_DATA_WRITTEN_CODE: // impossible
      CHECK_SPACE(pos, end, 4);
      data_written= uint4korr(pos);
      pos+= 4;
      break;
    case Q_INVOKER:
    {
      CHECK_SPACE(pos, end, 1);
      user.length= *pos++;
      CHECK_SPACE(pos, end, user.length);
      user.str= (char *)pos;
      pos+= user.length;

      CHECK_SPACE(pos, end, 1);
      host.length= *pos++;
      CHECK_SPACE(pos, end, host.length);
      host.str= (char *)pos;
      pos+= host.length;
      break;
    }
    case Q_HRNOW:
    {
      CHECK_SPACE(pos, end, 3);
      when_sec_part= uint3korr(pos);
      pos+= 3;
      break;
    }
   case Q_XID:
    {
      CHECK_SPACE(pos, end, 8);
      xid= uint8korr(pos);
      pos+= 8;
      break;
    }
    case Q_GTID_FLAGS3:
    {
      CHECK_SPACE(pos, end, 1);
      gtid_flags_extra= *pos++;
      if (gtid_flags_extra & (Gtid_log_event::FL_COMMIT_ALTER_E1 |
                              Gtid_log_event::FL_ROLLBACK_ALTER_E1))
      {
        CHECK_SPACE(pos, end, 8);
        sa_seq_no = uint8korr(pos);
        pos+= 8;
      }
      break;
    }
    case Q_DUMMY:
    {
      /*
        At some point, this query event was translated from a GTID event, with
        these Q_DUMMY bytes added to pad the end of the header. We can skip the
        rest of processing these vars. Note this is a separate case from the
        default to avoid the DBUG_PRINT of an unknown status var.
      */
      pos= (const uchar*) end;
      break;
    }
    default:
      /* That's why you must write status vars in growing order of code */
      DBUG_PRINT("info",("Query_log_event has unknown status vars (first has\
 code: %u), skipping the rest of them", (uint) *(pos-1)));
      pos= (const uchar*) end;                         // Break loop
    }
  }

#if !defined(MYSQL_CLIENT)
  if (description_event->server_version_split.kind ==
      Format_description_log_event::master_version_split::KIND_MYSQL)
  {
    // Handle MariaDB/MySQL incompatible sql_mode bits
    sql_mode_t mysql_sql_mode= sql_mode;
    sql_mode&= MODE_MASK_MYSQL_COMPATIBLE; // Unset MySQL specific bits

    /*
      sql_mode flags related to fraction second rounding/truncation
      have opposite meaning in MySQL vs MariaDB.
      MySQL:
       - rounds fractional seconds by default
       - truncates if TIME_TRUNCATE_FRACTIONAL is set
      MariaDB:
       - truncates fractional seconds by default
       - rounds if TIME_ROUND_FRACTIONAL is set
    */
    if (description_event->server_version_split >= fsp_version_split_mysql &&
       !(mysql_sql_mode & MODE_MYSQL80_TIME_TRUNCATE_FRACTIONAL))
      sql_mode|= MODE_TIME_ROUND_FRACTIONAL;
  }
#endif

  /**
    Layout for the data buffer is as follows
    +--------+-----------+------+------+---------+----+-------+
    | catlog | time_zone | user | host | db name | \0 | Query |
    +--------+-----------+------+------+---------+----+-------+

    To support the query cache we append the following buffer to the above
    +-------+---------------------------------------+-------+
    |db len | uninitialized space of size of db len | FLAGS |
    +-------+---------------------------------------+-------+

    The area of buffer starting from Query field all the way to the end belongs
    to the Query buffer and its structure is described in alloc_query() in
    sql_parse.cc
    */

#if !defined(MYSQL_CLIENT)
  if (!(start= data_buf= (Log_event::Byte*) my_malloc(PSI_INSTRUMENT_ME,
                                                       catalog_len + 1
                                                    +  time_zone_len + 1
                                                    +  user.length + 1
                                                    +  host.length + 1
                                                    +  data_len + 1
                                                    +  sizeof(size_t)//for db_len
                                                    +  db_len + 1
                                                    +  QUERY_CACHE_DB_LENGTH_SIZE
                                                    +  QUERY_CACHE_FLAGS_SIZE,
                                                       MYF(MY_WME))))
#else
  if (!(start= data_buf= (Log_event::Byte*) my_malloc(PSI_INSTRUMENT_ME,
                                                       catalog_len + 1
                                                    +  time_zone_len + 1
                                                    +  user.length + 1
                                                    +  host.length + 1
                                                    +  data_len + 1,
                                                       MYF(MY_WME))))
#endif
      DBUG_VOID_RETURN;
  if (catalog_len)                                  // If catalog is given
  {
    /**
      @todo we should clean up and do only copy_str_and_move; it
      works for both cases.  Then we can remove the catalog_nz
      flag. /sven
    */
    if (likely(catalog_nz)) // true except if event comes from 5.0.0|1|2|3.
      copy_str_and_move(&catalog, &start, catalog_len);
    else
    {
      memcpy(start, catalog, catalog_len+1); // copy end 0
      catalog= (const char *)start;
      start+= catalog_len+1;
    }
  }
  if (time_zone_len)
    copy_str_and_move(&time_zone_str, &start, time_zone_len);

  if (user.length)
  {
    copy_str_and_move(&user.str, &start, user.length);
  }
  else
  {
    user.str= (char*) start;
    *(start++)= 0;
  }

  if (host.length)
    copy_str_and_move(&host.str, &start, host.length);
  else
  {
    host.str= (char*) start;
    *(start++)= 0;
  }

  /**
    if time_zone_len or catalog_len are 0, then time_zone and catalog
    are uninitialized at this point.  shouldn't they point to the
    zero-length null-terminated strings we allocated space for in the
    my_alloc call above? /sven
  */

  /* A 2nd variable part; this is common to all versions */ 
  memcpy((char*) start, end, data_len);          // Copy db and query
  start[data_len]= '\0';              // End query with \0 (For safetly)
  db= (char *)start;
  query= (char *)(start + db_len + 1);
  q_len= data_len - db_len -1;

  if (data_len && (data_len < db_len ||
                   data_len < q_len ||
                   data_len != (db_len + q_len + 1)))
  {
    q_len= 0;
    query= NULL;
    DBUG_VOID_RETURN;
  }

  uint32 max_length= uint32(event_len - ((end + db_len + 1) -
                                         (buf - common_header_len)));
  if (q_len != max_length ||
      (event_len < uint((end + db_len + 1) - (buf - common_header_len))))
  {
    q_len= 0;
    query= NULL;
    DBUG_VOID_RETURN;
  }
  /**
    Append the db length at the end of the buffer. This will be used by
    Query_cache::send_result_to_client() in case the query cache is On.
   */
#if !defined(MYSQL_CLIENT)
  size_t db_length= (size_t)db_len;
  memcpy(start + data_len + 1, &db_length, sizeof(size_t));
#endif
  DBUG_VOID_RETURN;
}


/*
  Get the time when the event had been executed on the master.
  This works for both query events and load data events.
*/

#if Q_EXEC_TIME_OFFSET != L_EXEC_TIME_OFFSET
#error "Q_EXEC_TIME_OFFSET is not same as L_EXEC_TIME_OFFSET"
#endif

time_t
query_event_get_end_time(const uchar *buf,
                         const Format_description_log_event *description_event)
{
  time_t when;
  DBUG_ASSERT(LOG_EVENT_IS_QUERY((Log_event_type) buf[EVENT_TYPE_OFFSET]) ||
              LOG_EVENT_IS_LOAD_DATA((Log_event_type) buf[EVENT_TYPE_OFFSET]));
  when= uint4korr(buf);
  buf+= description_event->common_header_len;
  return when + uint4korr(buf + Q_EXEC_TIME_OFFSET);
}


Query_compressed_log_event::Query_compressed_log_event(const uchar *buf,
      uint event_len,
      const Format_description_log_event
      *description_event,
      Log_event_type event_type)
      :Query_log_event(buf, event_len, description_event, event_type),
       query_buf(NULL)
{
  if (query)
  {
    uint32 un_len= binlog_get_uncompress_len((uchar*) query);
    if (unlikely(!un_len || un_len > MAX_MAX_ALLOWED_PACKET))
    {
      query= 0;
      return;
    }

    /* Reserve one byte for '\0' */
    query_buf= (Log_event::Byte*) my_malloc(PSI_INSTRUMENT_ME,
                                            ALIGN_SIZE(un_len + 1), MYF(MY_WME));
    if (query_buf && !binlog_buf_uncompress((uchar*) query, (uchar *) query_buf,
                                            q_len, &un_len))
    {
      query_buf[un_len]= 0;
      query= (char*) query_buf;
      q_len= un_len;
    }
    else
    {
      query= 0;
    }
  }
}


/*
  Replace a binlog event read into a packet with a dummy event. Either a
  Query_log_event that has just a comment, or if that will not fit in the
  space used for the event to be replaced, then a NULL user_var event.

  This is used when sending binlog data to a slave which does not understand
  this particular event and which is too old to support informational events
  or holes in the event stream.

  This allows to write such events into the binlog on the master and still be
  able to replicate against old slaves without them breaking.

  Clears the flag LOG_EVENT_THREAD_SPECIFIC_F and set LOG_EVENT_SUPPRESS_USE_F.
  Overwrites the type with QUERY_EVENT (or USER_VAR_EVENT), and replaces the
  body with a minimal query / NULL user var.

  Returns zero on success, -1 if error due to too little space in original
  event. A minimum of 25 bytes (19 bytes fixed header + 6 bytes in the body)
  is needed in any event to be replaced with a dummy event.
*/
int
Query_log_event::dummy_event(String *packet, ulong ev_offset,
                             enum_binlog_checksum_alg checksum_alg)
{
  uchar *p= (uchar *)packet->ptr() + ev_offset;
  size_t data_len= packet->length() - ev_offset;
  uint16 flags;
  static const size_t min_user_var_event_len=
    LOG_EVENT_HEADER_LEN + UV_NAME_LEN_SIZE + 1 + UV_VAL_IS_NULL; // 25
  static const size_t min_query_event_len=
    LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN + 1 + 1; // 34

  if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
    data_len-= BINLOG_CHECKSUM_LEN;
  else
    DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF ||
                checksum_alg == BINLOG_CHECKSUM_ALG_OFF);

  if (data_len < min_user_var_event_len)
    /* Cannot replace with dummy, event too short. */
    return -1;

  flags= uint2korr(p + FLAGS_OFFSET);
  flags&= ~LOG_EVENT_THREAD_SPECIFIC_F;
  flags|= LOG_EVENT_SUPPRESS_USE_F;
  int2store(p + FLAGS_OFFSET, flags);

  if (data_len < min_query_event_len)
  {
    /*
      Have to use dummy user_var event for such a short packet.

      This works, but the event will be considered part of an event group with
      the following event. So for example @@global.sql_slave_skip_counter=1
      will skip not only the dummy event, but also the immediately following
      event.

      We write a NULL user var with the name @`!dummyvar` (or as much
      as that as will fit within the size of the original event - so
      possibly just @`!`).
    */
    static const char var_name[]= "!dummyvar";
    size_t name_len= data_len - (min_user_var_event_len - 1);

    p[EVENT_TYPE_OFFSET]= USER_VAR_EVENT;
    int4store(p + LOG_EVENT_HEADER_LEN, name_len);
    memcpy(p + LOG_EVENT_HEADER_LEN + UV_NAME_LEN_SIZE, var_name, name_len);
    p[LOG_EVENT_HEADER_LEN + UV_NAME_LEN_SIZE + name_len]= 1; // indicates NULL
  }
  else
  {
    /*
      Use a dummy query event, just a comment.
    */
    static const char message[]=
      "# Dummy event replacing event type %u that slave cannot handle.";
    char buf[sizeof(message)+1];  /* +1, as %u can expand to 3 digits. */
    uchar old_type= p[EVENT_TYPE_OFFSET];
    uchar *q= p + LOG_EVENT_HEADER_LEN;
    size_t comment_len, len;

    p[EVENT_TYPE_OFFSET]= QUERY_EVENT;
    int4store(q + Q_THREAD_ID_OFFSET, 0);
    int4store(q + Q_EXEC_TIME_OFFSET, 0);
    q[Q_DB_LEN_OFFSET]= 0;
    int2store(q + Q_ERR_CODE_OFFSET, 0);
    int2store(q + Q_STATUS_VARS_LEN_OFFSET, 0);
    q[Q_DATA_OFFSET]= 0;                    /* Zero terminator for empty db */
    q+= Q_DATA_OFFSET + 1;
    len= my_snprintf(buf, sizeof(buf), message, old_type);
    comment_len= data_len - (min_query_event_len - 1);
    if (comment_len <= len)
      memcpy(q, buf, comment_len);
    else
    {
      memcpy(q, buf, len);
      memset(q+len, ' ', comment_len - len);
    }
  }

  if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
  {
    ha_checksum crc= my_checksum(0, p, data_len);
    int4store(p + data_len, crc);
  }
  return 0;
}

/*
  Replace an event (GTID event) with a BEGIN query event, to be compatible
  with an old slave.
*/
int
Query_log_event::begin_event(String *packet, ulong ev_offset,
                             enum_binlog_checksum_alg checksum_alg)
{
  uchar *p= (uchar *)packet->ptr() + ev_offset;
  uchar *q= p + LOG_EVENT_HEADER_LEN;
  size_t data_len= packet->length() - ev_offset;
  uint dummy_bytes;
  uint16 flags;

  if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
    data_len-= BINLOG_CHECKSUM_LEN;
  else
    DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF ||
                checksum_alg == BINLOG_CHECKSUM_ALG_OFF);


  flags= uint2korr(p + FLAGS_OFFSET);
  flags&= ~LOG_EVENT_THREAD_SPECIFIC_F;
  flags|= LOG_EVENT_SUPPRESS_USE_F;
  int2store(p + FLAGS_OFFSET, flags);

  p[EVENT_TYPE_OFFSET]= QUERY_EVENT;
  int4store(q + Q_THREAD_ID_OFFSET, 0);
  int4store(q + Q_EXEC_TIME_OFFSET, 0);
  q[Q_DB_LEN_OFFSET]= 0;
  int2store(q + Q_ERR_CODE_OFFSET, 0);

  /*
    If the allocated GTID event packet header is longer than the size of the
    standard BEGIN query event's, then we need to fill in everything else with
    "dummy" values. That is, old slaves won't recognize the meaning for the
    DUMMY value, and will skip the rest of the status vars section.
  */
  DBUG_ASSERT(data_len >= LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN);
  if (data_len < LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN)
    return 1;
  DBUG_ASSERT(std::numeric_limits<uint16>::max() >=
              (data_len - (LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN)));
  dummy_bytes=
      static_cast<uint>(data_len - (LOG_EVENT_HEADER_LEN + GTID_HEADER_LEN));
  int2store(q + Q_STATUS_VARS_LEN_OFFSET, dummy_bytes);
  bfill(&q[Q_DATA_OFFSET], dummy_bytes, Q_DUMMY);
  q[Q_DATA_OFFSET + dummy_bytes]= 0; /* Zero terminator for empty db */
  q+= Q_DATA_OFFSET + dummy_bytes + 1;

  memcpy(q, "BEGIN", 5);

  if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
  {
    ha_checksum crc= my_checksum(0, p, data_len);
    int4store(p + data_len, crc);
  }
  return 0;
}


/***************************************************************************
       Format_description_log_event methods
****************************************************************************/

/**
  Format_description_log_event 1st ctor.

    Ctor. Can be used to create the event to write to the binary log (when the
    server starts or when FLUSH LOGS), or to create artificial events to parse
    binlogs from MySQL 3.23 or 4.x.
    When in a client, only the 2nd use is possible.

  @param binlog_version         the binlog version for which we want to build
                                an event. Can be 1 (=MySQL 3.23), 3 (=4.0.x
                                x>=2 and 4.1) or 4 (MySQL 5.0). Note that the
                                old 4.0 (binlog version 2) is not supported;
                                it should not be used for replication with
                                5.0.
  @param server_ver             a string containing the server version.
*/

Format_description_log_event::
Format_description_log_event(uint8 binlog_ver, const char* server_ver,
                             enum_binlog_checksum_alg checksum_alg)
  :Log_event(), created(0), binlog_version(binlog_ver),
   dont_set_created(0), event_type_permutation(0),
   used_checksum_alg(checksum_alg)
{
  switch (binlog_version) {
  case 4: /* MySQL 5.0 */
    memcpy(server_version, ::server_version, ST_SERVER_VER_LEN);
    DBUG_EXECUTE_IF("pretend_version_50034_in_binlog",
                    strmov(server_version, "5.0.34"););
    common_header_len= LOG_EVENT_HEADER_LEN;
    number_of_event_types= LOG_EVENT_TYPES;
    /* we'll catch my_malloc() error in is_valid() */
    post_header_len=(uint8*) my_malloc(PSI_INSTRUMENT_ME,
                                       number_of_event_types*sizeof(uint8)
                                       + BINLOG_CHECKSUM_ALG_DESC_LEN,
                                       MYF(0));
    /*
      This long list of assignments is not beautiful, but I see no way to
      make it nicer, as the right members are #defines, not array members, so
      it's impossible to write a loop.
    */
    if (post_header_len)
    {
#ifndef DBUG_OFF
      // Allows us to sanity-check that all events initialized their
      // events (see the end of this 'if' block).
      memset(post_header_len, 255, number_of_event_types*sizeof(uint8));
#endif

      /* Note: all event types must explicitly fill in their lengths here. */
      post_header_len[START_EVENT_V3-1]= START_V3_HEADER_LEN;
      post_header_len[QUERY_EVENT-1]= QUERY_HEADER_LEN;
      post_header_len[STOP_EVENT-1]= STOP_HEADER_LEN;
      post_header_len[ROTATE_EVENT-1]= ROTATE_HEADER_LEN;
      post_header_len[INTVAR_EVENT-1]= INTVAR_HEADER_LEN;
      post_header_len[LOAD_EVENT-1]= LOAD_HEADER_LEN;
      post_header_len[SLAVE_EVENT-1]= SLAVE_HEADER_LEN;
      post_header_len[CREATE_FILE_EVENT-1]= CREATE_FILE_HEADER_LEN;
      post_header_len[APPEND_BLOCK_EVENT-1]= APPEND_BLOCK_HEADER_LEN;
      post_header_len[EXEC_LOAD_EVENT-1]= EXEC_LOAD_HEADER_LEN;
      post_header_len[DELETE_FILE_EVENT-1]= DELETE_FILE_HEADER_LEN;
      post_header_len[NEW_LOAD_EVENT-1]= NEW_LOAD_HEADER_LEN;
      post_header_len[RAND_EVENT-1]= RAND_HEADER_LEN;
      post_header_len[USER_VAR_EVENT-1]= USER_VAR_HEADER_LEN;
      post_header_len[FORMAT_DESCRIPTION_EVENT-1]= FORMAT_DESCRIPTION_HEADER_LEN;
      post_header_len[XID_EVENT-1]= XID_HEADER_LEN;
      post_header_len[XA_PREPARE_LOG_EVENT-1]= XA_PREPARE_HEADER_LEN;
      post_header_len[BEGIN_LOAD_QUERY_EVENT-1]= BEGIN_LOAD_QUERY_HEADER_LEN;
      post_header_len[EXECUTE_LOAD_QUERY_EVENT-1]= EXECUTE_LOAD_QUERY_HEADER_LEN;
      /*
        The PRE_GA events are never be written to any binlog, but
        their lengths are included in Format_description_log_event.
        Hence, we need to be assign some value here, to avoid reading
        uninitialized memory when the array is written to disk.
      */
      post_header_len[PRE_GA_WRITE_ROWS_EVENT-1]= 0;
      post_header_len[PRE_GA_UPDATE_ROWS_EVENT-1]= 0;
      post_header_len[PRE_GA_DELETE_ROWS_EVENT-1]= 0;

      post_header_len[TABLE_MAP_EVENT-1]=       TABLE_MAP_HEADER_LEN;
      post_header_len[WRITE_ROWS_EVENT_V1-1]=   ROWS_HEADER_LEN_V1;
      post_header_len[UPDATE_ROWS_EVENT_V1-1]=  ROWS_HEADER_LEN_V1;
      post_header_len[DELETE_ROWS_EVENT_V1-1]=  ROWS_HEADER_LEN_V1;
      /*
        We here have the possibility to simulate a master of before we changed
        the table map id to be stored in 6 bytes: when it was stored in 4
        bytes (=> post_header_len was 6). This is used to test backward
        compatibility.
        This code can be removed after a few months (today is Dec 21st 2005),
        when we know that the 4-byte masters are not deployed anymore (check
        with Tomas Ulin first!), and the accompanying test (rpl_row_4_bytes)
        too.
      */
      DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master",
                      post_header_len[TABLE_MAP_EVENT-1]=
                      post_header_len[WRITE_ROWS_EVENT_V1-1]=
                      post_header_len[UPDATE_ROWS_EVENT_V1-1]=
                      post_header_len[DELETE_ROWS_EVENT_V1-1]= 6;);
      post_header_len[INCIDENT_EVENT-1]= INCIDENT_HEADER_LEN;
      post_header_len[HEARTBEAT_LOG_EVENT-1]= 0;
      post_header_len[IGNORABLE_LOG_EVENT-1]= 0;
      post_header_len[ROWS_QUERY_LOG_EVENT-1]= 0;
      post_header_len[GTID_LOG_EVENT-1]= 0;
      post_header_len[ANONYMOUS_GTID_LOG_EVENT-1]= 0;
      post_header_len[PREVIOUS_GTIDS_LOG_EVENT-1]= 0;
      post_header_len[TRANSACTION_CONTEXT_EVENT-1]= 0;
      post_header_len[VIEW_CHANGE_EVENT-1]= 0;
      post_header_len[XA_PREPARE_LOG_EVENT-1]= 0;
      post_header_len[PARTIAL_UPDATE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2;
      post_header_len[TRANSACTION_PAYLOAD_EVENT-1]= ROWS_HEADER_LEN_V2;
      post_header_len[HEARTBEAT_LOG_EVENT_V2-1]= ROWS_HEADER_LEN_V2;
      post_header_len[WRITE_ROWS_EVENT-1]=  ROWS_HEADER_LEN_V2;
      post_header_len[UPDATE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2;
      post_header_len[DELETE_ROWS_EVENT-1]= ROWS_HEADER_LEN_V2;

      // Set header length of the reserved events to 0
      memset(post_header_len + MYSQL_EVENTS_END - 1, 0,
             (MARIA_EVENTS_BEGIN - MYSQL_EVENTS_END)*sizeof(uint8));

      // Set header lengths of Maria events
      post_header_len[ANNOTATE_ROWS_EVENT-1]= ANNOTATE_ROWS_HEADER_LEN;
      post_header_len[BINLOG_CHECKPOINT_EVENT-1]=
        BINLOG_CHECKPOINT_HEADER_LEN;
      post_header_len[GTID_EVENT-1]= GTID_HEADER_LEN;
      post_header_len[GTID_LIST_EVENT-1]= GTID_LIST_HEADER_LEN;
      post_header_len[START_ENCRYPTION_EVENT-1]= START_ENCRYPTION_HEADER_LEN;

      //compressed event
      post_header_len[QUERY_COMPRESSED_EVENT-1]= QUERY_HEADER_LEN;
      post_header_len[WRITE_ROWS_COMPRESSED_EVENT-1]=   ROWS_HEADER_LEN_V2;
      post_header_len[UPDATE_ROWS_COMPRESSED_EVENT-1]=  ROWS_HEADER_LEN_V2;
      post_header_len[DELETE_ROWS_COMPRESSED_EVENT-1]=  ROWS_HEADER_LEN_V2;
      post_header_len[WRITE_ROWS_COMPRESSED_EVENT_V1-1]=   ROWS_HEADER_LEN_V1;
      post_header_len[UPDATE_ROWS_COMPRESSED_EVENT_V1-1]=  ROWS_HEADER_LEN_V1;
      post_header_len[DELETE_ROWS_COMPRESSED_EVENT_V1-1]=  ROWS_HEADER_LEN_V1;
      post_header_len[PARTIAL_ROW_DATA_EVENT-1]=  PARTIAL_ROWS_HEADER_LEN;

      // Sanity-check that all post header lengths are initialized.
      int i;
      for (i=0; i<number_of_event_types; i++)
        DBUG_ASSERT(post_header_len[i] != 255);
    }
    break;

  case 1: /* 3.23 */
  case 3: /* 4.0.x x>=2 */
  default: /* Includes binlog version 2 i.e. 4.0.x x<=1 */
    post_header_len= 0; /* will make is_valid() fail */
    break;
  }
  calc_server_version_split();
  deduct_options_written_to_bin_log();
  reset_crypto();
}


/**
  The problem with this constructor is that the fixed header may have a
  length different from this version, but we don't know this length as we
  have not read the Format_description_log_event which says it, yet. This
  length is in the post-header of the event, but we don't know where the
  post-header starts.

  So this type of event HAS to:
  - either have the header's length at the beginning (in the header, at a
  fixed position which will never be changed), not in the post-header. That
  would make the header be "shifted" compared to other events.
  - or have a header of size LOG_EVENT_MINIMAL_HEADER_LEN (19), in all future
  versions, so that we know for sure.

  I (Guilhem) chose the 2nd solution. Rotate has the same constraint (because
  it is sent before Format_description_log_event).
*/

Format_description_log_event::
Format_description_log_event(const uchar *buf, uint event_len,
                             const Format_description_log_event*
                             description_event)
  :Log_event(buf, description_event), binlog_version(BINLOG_VERSION),
   common_header_len(0), post_header_len(NULL), event_type_permutation(0)
{
  DBUG_ENTER("Format_description_log_event::Format_description_log_event(char*,...)");
  used_checksum_alg= BINLOG_CHECKSUM_ALG_UNDEF;
  if (unlikely(
      event_len < LOG_EVENT_MINIMAL_HEADER_LEN + ST_POST_HEADER_LEN_OFFSET))
  {
    server_version[0]= 0;
    DBUG_VOID_RETURN;
  }
  buf+= LOG_EVENT_MINIMAL_HEADER_LEN;
  binlog_version= uint2korr(buf+ST_BINLOG_VER_OFFSET);
  memcpy(server_version, buf+ST_SERVER_VER_OFFSET, ST_SERVER_VER_LEN);
  // prevent overrun if log is corrupted on disk
  server_version[ST_SERVER_VER_LEN-1]= 0;
  created= uint4korr(buf+ST_CREATED_OFFSET);
  dont_set_created= 1;

  if (unlikely(server_version[0] == 0))
    DBUG_VOID_RETURN; /* sanity check */
  if (unlikely(
      (common_header_len=buf[ST_COMMON_HEADER_LEN_OFFSET]) < LOG_EVENT_MINIMAL_HEADER_LEN))
    DBUG_VOID_RETURN; /* sanity check */
  number_of_event_types=
    event_len - (LOG_EVENT_MINIMAL_HEADER_LEN + ST_POST_HEADER_LEN_OFFSET);
  DBUG_PRINT("info", ("common_header_len=%d number_of_event_types=%d",
                      common_header_len, number_of_event_types));
  /* If alloc fails, we'll detect it in is_valid() */

  calc_server_version_split();
  buf+= ST_POST_HEADER_LEN_OFFSET;
  if (!is_version_before_checksum(&server_version_split))
  {
    /* the last bytes are the checksum alg desc */
    if (unlikely(number_of_event_types < BINLOG_CHECKSUM_ALG_DESC_LEN))
      DBUG_VOID_RETURN; /* sanity check: But there are no last bytes. */
    number_of_event_types -= BINLOG_CHECKSUM_ALG_DESC_LEN;
    used_checksum_alg= (enum_binlog_checksum_alg)buf[number_of_event_types];
  }
  else
  {
    used_checksum_alg= BINLOG_CHECKSUM_ALG_OFF;
  }
  post_header_len= (uint8*) my_memdup(PSI_INSTRUMENT_ME,
                                      buf,
                                      number_of_event_types*
                                      sizeof(*post_header_len),
                                      MYF(0));
  deduct_options_written_to_bin_log();
  reset_crypto();

  DBUG_VOID_RETURN;
}

bool Format_description_log_event::start_decryption(Start_encryption_log_event* sele)
{
  DBUG_ASSERT(crypto_data.scheme == 0);

  if (!sele->is_valid())
    return 1;

  memcpy(crypto_data.nonce, sele->nonce, BINLOG_NONCE_LENGTH);
  return crypto_data.init(sele->crypto_scheme, sele->key_version);
}


Version::Version(const char *version, const char **endptr)
{
  const char *p= version;
  ulong number;
  for (uint i= 0; i<=2; i++)
  {
    char *r;
    number= strtoul(p, &r, 10);
    /*
      It is an invalid version if any version number greater than 255 or
      first number is not followed by '.'.
    */
    if (number < 256 && (*r == '.' || i != 0))
      m_ver[i]= (uchar) number;
    else
    {
      *this= Version();
      break;
    }

    p= r;
    if (*r == '.')
      p++; // skip the dot
  }
  endptr[0]= p;
}


Format_description_log_event::
  master_version_split::master_version_split(const char *version)
{
  const char *p;
  static_cast<Version*>(this)[0]= Version(version, &p);
  if (strstr(p, "MariaDB") != 0 || strstr(p, "-maria-") != 0)
    kind= KIND_MARIADB;
  else
    kind= KIND_MYSQL;
}


/**
   Splits the event's 'server_version' string into three numeric pieces stored
   into 'server_version_split':
   X.Y.Zabc (X,Y,Z numbers, a not a digit) -> {X,Y,Z}
   X.Yabc -> {X,Y,0}
   'server_version_split' is then used for lookups to find if the server which
   created this event has some known bug.
*/
void Format_description_log_event::calc_server_version_split()
{
  server_version_split= master_version_split(server_version);

  DBUG_PRINT("info",("Format_description_log_event::server_version_split:"
                     " '%s' %d %d %d", server_version,
                     server_version_split[0],
                     server_version_split[1], server_version_split[2]));
}


void Format_description_log_event::deduct_options_written_to_bin_log()
{
  options_written_to_bin_log= OPTION_AUTO_IS_NULL | OPTION_NOT_AUTOCOMMIT |
              OPTION_NO_FOREIGN_KEY_CHECKS | OPTION_RELAXED_UNIQUE_CHECKS |
              OPTION_INSERT_HISTORY | OPTION_NO_CHECK_CONSTRAINT_CHECKS;
  if (!server_version_split.version_is_valid() ||
      server_version_split.kind == master_version_split::KIND_MYSQL ||
      server_version_split < Version(10,5,2))
    return;
  options_written_to_bin_log|= OPTION_IF_EXISTS;
  if (server_version_split[0] == 10)
  {
    const static char v[10]={99,99,99,99,99,17,9,5,4,2};
    if (server_version_split[1] < 10 &&
        server_version_split[2] < v[server_version_split[1]])
      return;
  }
  options_written_to_bin_log|= OPTION_EXPLICIT_DEF_TIMESTAMP;

  DBUG_ASSERT(options_written_to_bin_log == OPTIONS_WRITTEN_TO_BIN_LOG);
}

/**
   @return TRUE is the event's version is earlier than one that introduced
   the replication event checksum. FALSE otherwise.
*/
bool
Format_description_log_event::is_version_before_checksum(const master_version_split
                                                         *version_split)
{
  return *version_split <
    (version_split->kind == master_version_split::KIND_MARIADB ?
     checksum_version_split_mariadb : checksum_version_split_mysql);
}

/**
   @param buf buffer holding serialized FD event including the 4-byte checksum
   @param len length of the event buf
   @param alg output the version-safe checksum alg descriptor where zero
            designates no checksum, 255 - the orginator is
            checksum-unaware (effectively no checksum) and the actual
            [1-254] range alg descriptor.
   @return whether this is an invalid FD event
*/
bool get_checksum_alg(const uchar *buf, size_t len,
                      enum_binlog_checksum_alg *alg)
{
  constexpr ptrdiff_t POST_HEADER_LEN_OFFSET=
    LOG_EVENT_MINIMAL_HEADER_LEN + ST_POST_HEADER_LEN_OFFSET;
  char version[ST_SERVER_VER_LEN];

  DBUG_ENTER("get_checksum_alg");
  DBUG_ASSERT(buf[EVENT_TYPE_OFFSET] == FORMAT_DESCRIPTION_EVENT);

  if (unlikely(len < POST_HEADER_LEN_OFFSET))
    DBUG_RETURN(true);
  memcpy(version,
         buf + LOG_EVENT_MINIMAL_HEADER_LEN + ST_SERVER_VER_OFFSET,
         ST_SERVER_VER_LEN);
  version[ST_SERVER_VER_LEN - 1]= 0;
  
  Format_description_log_event::master_version_split version_split(version);
  if (Format_description_log_event::is_version_before_checksum(&version_split))
    *alg= BINLOG_CHECKSUM_ALG_UNDEF;
  else
  {
    /*
      len >= POST_HEADER_LEN_OFFSET >
      BINLOG_CHECKSUM_LEN + BINLOG_CHECKSUM_ALG_DESC_LEN
    */
    size_t checksum_alg_offset=
      len - BINLOG_CHECKSUM_LEN - BINLOG_CHECKSUM_ALG_DESC_LEN;
    if (unlikely(checksum_alg_offset < POST_HEADER_LEN_OFFSET))
      DBUG_RETURN(true);
    *alg= static_cast<enum_binlog_checksum_alg>(buf[checksum_alg_offset]);
  }
  DBUG_ASSERT(*alg == BINLOG_CHECKSUM_ALG_OFF ||
              *alg == BINLOG_CHECKSUM_ALG_UNDEF ||
              *alg == BINLOG_CHECKSUM_ALG_CRC32);
  DBUG_RETURN(false);
}

Start_encryption_log_event::
Start_encryption_log_event(const uchar *buf, uint event_len,
                           const Format_description_log_event* description_event)
  :Log_event(buf, description_event)
{
  if ((int)event_len ==
      LOG_EVENT_MINIMAL_HEADER_LEN + Start_encryption_log_event::get_data_size())
  {
    buf+= LOG_EVENT_MINIMAL_HEADER_LEN;
    crypto_scheme= *buf;
    key_version= uint4korr(buf + BINLOG_CRYPTO_SCHEME_LENGTH);
    memcpy(nonce,
           buf + BINLOG_CRYPTO_SCHEME_LENGTH + BINLOG_KEY_VERSION_LENGTH,
           BINLOG_NONCE_LENGTH);
  }
  else
    crypto_scheme= ~0; // invalid
}



/**************************************************************************
  Rotate_log_event methods
**************************************************************************/

Rotate_log_event::Rotate_log_event(const uchar *buf, uint event_len,
                                   const Format_description_log_event*
                                   description_event)
  :Log_event(buf, description_event) ,new_log_ident(0), flags(DUP_NAME)
{
  DBUG_ENTER("Rotate_log_event::Rotate_log_event(char*,...)");
  // The caller will ensure that event_len is what we have at EVENT_LEN_OFFSET
  uint8 post_header_len= description_event->post_header_len[ROTATE_EVENT-1];
  uint ident_offset;
  if (event_len < (uint)(LOG_EVENT_MINIMAL_HEADER_LEN + post_header_len))
    DBUG_VOID_RETURN;
  buf+= LOG_EVENT_MINIMAL_HEADER_LEN;
  pos= post_header_len ? uint8korr(buf + R_POS_OFFSET) : 4;
  ident_len= (uint)(event_len - (LOG_EVENT_MINIMAL_HEADER_LEN + post_header_len));
  ident_offset= post_header_len;
  set_if_smaller(ident_len,FN_REFLEN-1);
  new_log_ident= my_strndup(PSI_INSTRUMENT_ME, (char*) buf + ident_offset,
                            (uint) ident_len, MYF(MY_WME));
  DBUG_PRINT("debug", ("new_log_ident: '%s'", new_log_ident));
  DBUG_VOID_RETURN;
}


/**************************************************************************
  Binlog_checkpoint_log_event methods
**************************************************************************/

Binlog_checkpoint_log_event::Binlog_checkpoint_log_event(
       const uchar *buf, uint event_len,
       const Format_description_log_event *description_event)
  :Log_event(buf, description_event), binlog_file_name(0)
{
  uint8 header_size= description_event->common_header_len;
  uint8 post_header_len=
    description_event->post_header_len[BINLOG_CHECKPOINT_EVENT-1];
  if (event_len < (uint) header_size + (uint) post_header_len ||
      post_header_len < BINLOG_CHECKPOINT_HEADER_LEN)
    return;
  buf+= header_size;
  /* See uint4korr and int4store below */
  compile_time_assert(BINLOG_CHECKPOINT_HEADER_LEN == 4);
  binlog_file_len= uint4korr(buf);
  if (event_len - (header_size + post_header_len) < binlog_file_len)
    return;
  binlog_file_name= my_strndup(PSI_INSTRUMENT_ME, (char*) buf + post_header_len,
                               binlog_file_len, MYF(MY_WME));
  return;
}


/**************************************************************************
        Global transaction ID stuff
**************************************************************************/

Gtid_log_event::Gtid_log_event(const uchar *buf, uint event_len,
                               const Format_description_log_event
                               *description_event)
  : Log_event(buf, description_event), seq_no(0), commit_id(0),
    flags_extra(0), extra_engines(0), thread_id(0)
{
  uint8 header_size= description_event->common_header_len;
  uint8 post_header_len= description_event->post_header_len[GTID_EVENT-1];
  const uchar *buf_0= buf;
  if (event_len < (uint) header_size + (uint) post_header_len ||
      post_header_len < GTID_HEADER_LEN)
    return;

  buf+= header_size;
  seq_no= uint8korr(buf);
  buf+= 8;
  domain_id= uint4korr(buf);
  buf+= 4;
  flags2= *(buf++);
  if (flags2 & FL_GROUP_COMMIT_ID)
  {
    if (event_len < (uint)header_size + GTID_HEADER_LEN + 2)
    {
      seq_no= 0;                                // So is_valid() returns false
      return;
    }
    commit_id= uint8korr(buf);
    buf+= 8;
  }
  if (flags2 & (FL_PREPARED_XA | FL_COMPLETED_XA))
  {
    if (event_len < static_cast<uint>(buf - buf_0) + 6)
    {
      seq_no= 0;
      return;
    }
    xid.formatID= uint4korr(buf);
    buf+= 4;

    xid.gtrid_length= (long) buf[0];
    xid.bqual_length= (long) buf[1];
    buf+= 2;

    long data_length= xid.bqual_length + xid.gtrid_length;
    if (event_len < static_cast<uint>(buf - buf_0) + data_length ||
        xid.gtrid_length > MAXGTRIDSIZE || xid.bqual_length > MAXBQUALSIZE)
    {
      xid.formatID= -1;
      seq_no= 0;
      return;
    }
    memcpy(xid.data, buf, data_length);
    buf+= data_length;
  }

  /* the extra flags check and actions */
  if (static_cast<uint>(buf - buf_0) < event_len)
  {
    flags_extra= *buf++;
    /*
      extra engines flags presence is identified by non-zero byte value
      at this point
    */
    if (flags_extra & FL_EXTRA_MULTI_ENGINE_E1)
    {
      if (event_len < static_cast<uint>(buf - buf_0) + 1)
      {
        seq_no= 0;
        return;
      }
      extra_engines= *buf++;

      DBUG_ASSERT(extra_engines > 0);
    }
    if (flags_extra & (FL_COMMIT_ALTER_E1 | FL_ROLLBACK_ALTER_E1))
    {
      if (event_len < static_cast<uint>(buf - buf_0) + 8)
      {
        seq_no= 0;
        return;
      }
      sa_seq_no= uint8korr(buf);
      buf+= 8;
    }

    if (flags_extra & FL_EXTRA_THREAD_ID &&
        static_cast<uint>(buf - buf_0) <= event_len + 4)
    {
      thread_id= uint4korr(buf);
      buf+= 4;
    }
  }
  /*
    the strict '<' part of the assert corresponds to extra zero-padded
    trailing bytes,
  */
  DBUG_ASSERT(static_cast<uint>(buf - buf_0) <= event_len);
  /* and the last of them is tested. */
#ifdef MYSQL_SERVER
#ifdef WITH_WSREP
  if (!WSREP_ON)
#endif
#endif
  DBUG_ASSERT(static_cast<uint>(buf - buf_0) == event_len ||
              buf_0[event_len - 1] == 0);
}

int compare_glle_gtids(const void * _gtid1, const void *_gtid2)
{
  rpl_gtid *gtid1= (rpl_gtid *) _gtid1;
  rpl_gtid *gtid2= (rpl_gtid *) _gtid2;

  int ret;
  if (*gtid1 < *gtid2)
    ret= -1;
  else if (*gtid1 > *gtid2)
    ret= 1;
  else
    ret= 0;
  return ret;
}

/* GTID list. */

Gtid_list_log_event::Gtid_list_log_event(const uchar *buf, uint event_len,
                                         const Format_description_log_event
                                         *description_event)
  : Log_event(buf, description_event), count(0), list(0), sub_id_list(0)
{
  uint32 i;
  uint32 val;
  uint8 header_size= description_event->common_header_len;
  uint8 post_header_len= description_event->post_header_len[GTID_LIST_EVENT-1];
  if (event_len < (uint) header_size + (uint) post_header_len ||
      post_header_len < GTID_LIST_HEADER_LEN)
    return;

  buf+= header_size;
  val= uint4korr(buf);
  count= val & ((1<<28)-1);
  gl_flags= val & ((uint32)0xf << 28);
  buf+= 4;
  if (event_len - (header_size + post_header_len) < count*element_size ||
      (!(list= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME,
                            count*sizeof(*list) + (count == 0), MYF(MY_WME)))))
    return;

  for (i= 0; i < count; ++i)
  {
    list[i].domain_id= uint4korr(buf);
    buf+= 4;
    list[i].server_id= uint4korr(buf);
    buf+= 4;
    list[i].seq_no= uint8korr(buf);
    buf+= 8;
  }

#if defined(HAVE_REPLICATION) && !defined(MYSQL_CLIENT)
  if ((gl_flags & FLAG_IGN_GTIDS))
  {
    uint32 i;
    if (!(sub_id_list= (uint64 *)my_malloc(PSI_INSTRUMENT_ME,
                                           count*sizeof(uint64), MYF(MY_WME))))
    {
      my_free(list);
      list= NULL;
      return;
    }
    for (i= 0; i < count; ++i)
    {
      if (!(sub_id_list[i]=
            rpl_global_gtid_slave_state->next_sub_id(list[i].domain_id)))
      {
        my_free(list);
        my_free(sub_id_list);
        list= NULL;
        sub_id_list= NULL;
        return;
      }
    }
  }
#endif
}


/*
  Used to record gtid_list event while sending binlog to slave, without having to
  fully construct the event object.
*/
bool
Gtid_list_log_event::peek(const char *event_start, size_t event_len,
                          enum_binlog_checksum_alg checksum_alg,
                          rpl_gtid **out_gtid_list, uint32 *out_list_len,
                          const Format_description_log_event *fdev)
{
  const char *p;
  uint32 count_field, count;
  rpl_gtid *gtid_list;

  if (checksum_alg == BINLOG_CHECKSUM_ALG_CRC32)
  {
    if (event_len > BINLOG_CHECKSUM_LEN)
      event_len-= BINLOG_CHECKSUM_LEN;
    else
      event_len= 0;
  }
  else
    DBUG_ASSERT(checksum_alg == BINLOG_CHECKSUM_ALG_UNDEF ||
                checksum_alg == BINLOG_CHECKSUM_ALG_OFF);

  if (event_len < (uint32)fdev->common_header_len + GTID_LIST_HEADER_LEN)
    return true;
  p= event_start + fdev->common_header_len;
  count_field= uint4korr(p);
  p+= 4;
  count= count_field & ((1<<28)-1);
  if (event_len < (uint32)fdev->common_header_len + GTID_LIST_HEADER_LEN +
      element_size * count)
    return true;
  if (!(gtid_list= (rpl_gtid *)my_malloc(PSI_INSTRUMENT_ME,
                          sizeof(rpl_gtid)*count + (count == 0), MYF(MY_WME))))
    return true;
  *out_gtid_list= gtid_list;
  *out_list_len= count;
  while (count--)
  {
    gtid_list->domain_id= uint4korr(p);
    p+= 4;
    gtid_list->server_id= uint4korr(p);
    p+= 4;
    gtid_list->seq_no= uint8korr(p);
    p+= 8;
    ++gtid_list;
  }

  return false;
}


/**************************************************************************
 Intvar_log_event methods
**************************************************************************/

/*
  Intvar_log_event::Intvar_log_event()
*/

Intvar_log_event::Intvar_log_event(const uchar *buf,
                                   const Format_description_log_event* description_event)
  :Log_event(buf, description_event)
{
  /* The Post-Header is empty. The Variable Data part begins immediately. */
  buf+= description_event->common_header_len +
    description_event->post_header_len[INTVAR_EVENT-1];
  type= buf[I_TYPE_OFFSET];
  val= uint8korr(buf+I_VAL_OFFSET);
}


/*
  Intvar_log_event::get_var_type_name()
*/

const char* Intvar_log_event::get_var_type_name()
{
  switch(type) {
  case LAST_INSERT_ID_EVENT: return "LAST_INSERT_ID";
  case INSERT_ID_EVENT: return "INSERT_ID";
  default: /* impossible */ return "UNKNOWN";
  }
}


/**************************************************************************
  Rand_log_event methods
**************************************************************************/

Rand_log_event::Rand_log_event(const uchar *buf,
                               const Format_description_log_event* description_event)
  :Log_event(buf, description_event)
{
  /* The Post-Header is empty. The Variable Data part begins immediately. */
  buf+= description_event->common_header_len +
    description_event->post_header_len[RAND_EVENT-1];
  seed1= uint8korr(buf+RAND_SEED1_OFFSET);
  seed2= uint8korr(buf+RAND_SEED2_OFFSET);
}


/**************************************************************************
  Xid_log_event methods
**************************************************************************/

/**
  @note
  It's ok not to use int8store here,
  as long as xid_t::set(ulonglong) and
  xid_t::get_my_xid doesn't do it either.
  We don't care about actual values of xids as long as
  identical numbers compare identically
*/

Xid_log_event::
Xid_log_event(const uchar *buf,
              const Format_description_log_event *description_event)
  :Xid_apply_log_event(buf, description_event)
{
  /* The Post-Header is empty. The Variable Data part begins immediately. */
  buf+= description_event->common_header_len +
    description_event->post_header_len[XID_EVENT-1];
  memcpy((char*) &xid, buf, sizeof(xid));
}

/**************************************************************************
  XA_prepare_log_event methods
**************************************************************************/
XA_prepare_log_event::
XA_prepare_log_event(const uchar *buf,
                     const Format_description_log_event *description_event)
  :Xid_apply_log_event(buf, description_event)
{
  buf+= description_event->common_header_len +
    description_event->post_header_len[XA_PREPARE_LOG_EVENT-1];
  one_phase= * (bool *) buf;
  buf+= 1;

  m_xid.formatID= uint4korr(buf);
  buf+= 4;
  m_xid.gtrid_length= uint4korr(buf);
  buf+= 4;
  // Todo: validity here and elsewhere checks to be replaced by MDEV-21839 fixes
  if (m_xid.gtrid_length <= 0 || m_xid.gtrid_length > MAXGTRIDSIZE)
  {
    m_xid.formatID= -1;
    return;
  }
  m_xid.bqual_length= uint4korr(buf);
  buf+= 4;
  if (m_xid.bqual_length < 0 || m_xid.bqual_length > MAXBQUALSIZE)
  {
    m_xid.formatID= -1;
    return;
  }
  DBUG_ASSERT(m_xid.gtrid_length + m_xid.bqual_length <= XIDDATASIZE);

  memcpy(m_xid.data, buf, m_xid.gtrid_length + m_xid.bqual_length);

  xid= NULL;
}


/**************************************************************************
  User_var_log_event methods
**************************************************************************/

bool Log_event_data_type::unpack_optional_attributes(const char *pos,
                                                     const char *end)

{
  for ( ; pos < end; )
  {
    switch (*pos) {
    case CHUNK_SIGNED:
      m_is_unsigned= false;
      pos++;
      continue;
    case CHUNK_UNSIGNED:
      m_is_unsigned= true;
      pos++;
      continue;
    case CHUNK_DATA_TYPE_NAME:
      {
        pos++;
        if (pos >= end)
          return true;
        uint length= (uchar) *pos++;
        if (pos + length > end)
          return true;
        m_data_type_name= {pos, length};
        pos+= length;
        continue;
      }
    default:
      break; // Unknown chunk
    }
  }
  return false;
}


User_var_log_event::
User_var_log_event(const uchar *buf, uint event_len,
                   const Format_description_log_event* description_event)
  :Log_event(buf, description_event)
#ifndef MYSQL_CLIENT
  , deferred(false), query_id(0)
#endif
{
  bool error= false;
  const uchar *const buf_start= buf;
  const char *buf_end= reinterpret_cast<const char*>(buf) + event_len;

  /* The Post-Header is empty. The Variable Data part begins immediately. */
  buf+= description_event->common_header_len +
    description_event->post_header_len[USER_VAR_EVENT-1];
  name_len= uint4korr(buf);
  /* Avoid reading out of buffer */
  if ((buf - buf_start) + UV_NAME_LEN_SIZE + name_len > event_len)
  {
    error= true;
    goto err;
  }

  name= (char *) buf + UV_NAME_LEN_SIZE;

  /*
    We don't know yet is_null value, so we must assume that name_len
    may have the bigger value possible, is_null= True and there is no
    payload for val, or even that name_len is 0.
  */
  if (name + name_len + UV_VAL_IS_NULL > (char*) buf_end)
  {
    error= true;
    goto err;
  }

  buf+= UV_NAME_LEN_SIZE + name_len;
  is_null= (bool) *buf;
  if (is_null)
  {
    val_len= 0;
    val= 0;  
  }
  else
  {
    val= (char *) (buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
                   UV_CHARSET_NUMBER_SIZE + UV_VAL_LEN_SIZE);

    if (val > (char*) buf_end)
    {
      error= true;
      goto err;
    }

    m_type= (Item_result) buf[UV_VAL_IS_NULL];
    m_charset_number= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE);
    val_len= uint4korr(buf + UV_VAL_IS_NULL + UV_VAL_TYPE_SIZE +
                       UV_CHARSET_NUMBER_SIZE);

    /**
      We need to check if this is from an old server
      that did not pack information for flags.
      We do this by checking if there are extra bytes
      after the packed value. If there are we take the
      extra byte and it's value is assumed to contain
      the flags value.

      Old events will not have this extra byte, thence,
      we keep m_is_unsigned==false.
    */
    const char *pos= val + val_len;
    if (pos > buf_end || unpack_optional_attributes(pos, buf_end))
    {
      error= true;
      goto err;
    }
  }

err:
  if (unlikely(error))
    name= 0;
}


/**************************************************************************
 Append_block_log_event methods
**************************************************************************/

/*
  Append_block_log_event ctor
*/

Append_block_log_event::
Append_block_log_event(const uchar *buf, uint len,
                       const Format_description_log_event* description_event)
  :Log_event(buf, description_event),block(0)
{
  DBUG_ENTER("Append_block_log_event::Append_block_log_event(char*,...)");
  uint8 common_header_len= description_event->common_header_len; 
  uint8 append_block_header_len=
    description_event->post_header_len[APPEND_BLOCK_EVENT-1];
  uint total_header_len= common_header_len+append_block_header_len;
  if (len < total_header_len)
    DBUG_VOID_RETURN;
  file_id= uint4korr(buf + common_header_len + AB_FILE_ID_OFFSET);
  block= const_cast<uchar*>(buf) + total_header_len;
  block_len= len - total_header_len;
  DBUG_VOID_RETURN;
}


/**************************************************************************
 Delete_file_log_event methods
**************************************************************************/

/*
  Delete_file_log_event ctor
*/

Delete_file_log_event::
Delete_file_log_event(const uchar *buf, uint len,
                      const Format_description_log_event* description_event)
  :Log_event(buf, description_event),file_id(0)
{
  uint8 common_header_len= description_event->common_header_len;
  uint8 delete_file_header_len= description_event->post_header_len[DELETE_FILE_EVENT-1];
  if (len < (uint)(common_header_len + delete_file_header_len))
    return;
  file_id= uint4korr(buf + common_header_len + DF_FILE_ID_OFFSET);
}


/**************************************************************************
 Begin_load_query_log_event methods
**************************************************************************/

Begin_load_query_log_event::
Begin_load_query_log_event(const uchar *buf, uint len,
                           const Format_description_log_event* desc_event)
  :Append_block_log_event(buf, len, desc_event)
{
}


/**************************************************************************
 Execute_load_query_log_event methods
**************************************************************************/


Execute_load_query_log_event::
Execute_load_query_log_event(const uchar *buf, uint event_len,
                             const Format_description_log_event* desc_event):
  Query_log_event(buf, event_len, desc_event, EXECUTE_LOAD_QUERY_EVENT),
  file_id(0), fn_pos_start(0), fn_pos_end(0)
{
  if (!Query_log_event::is_valid())
    return;

  buf+= desc_event->common_header_len;

  fn_pos_start= uint4korr(buf + ELQ_FN_POS_START_OFFSET);
  fn_pos_end= uint4korr(buf + ELQ_FN_POS_END_OFFSET);
  dup_handling= (enum_load_dup_handling)(*(buf + ELQ_DUP_HANDLING_OFFSET));

  if (fn_pos_start > q_len || fn_pos_end > q_len ||
      dup_handling > LOAD_DUP_REPLACE)
    return;

  file_id= uint4korr(buf + ELQ_FILE_ID_OFFSET);
}


ulong Execute_load_query_log_event::get_post_header_size_for_derived()
{
  return EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN;
}


/**************************************************************************
 sql_ex_info methods
**************************************************************************/

/*
  sql_ex_info::init()
*/

const uchar *sql_ex_info::init(const uchar *buf, const uchar *buf_end,
                              bool use_new_format)
{
  cached_new_format= use_new_format;
  if (use_new_format)
  {
    empty_flags=0;
    /*
      The code below assumes that buf will not disappear from
      under our feet during the lifetime of the event. This assumption
      holds true in the slave thread if the log is in new format, but is not
      the case when we have old format because we will be reusing net buffer
      to read the actual file before we write out the Create_file event.
    */
    if (read_str(&buf, buf_end, &field_term, &field_term_len) ||
        read_str(&buf, buf_end, &enclosed,   &enclosed_len) ||
        read_str(&buf, buf_end, &line_term,  &line_term_len) ||
        read_str(&buf, buf_end, &line_start, &line_start_len) ||
        read_str(&buf, buf_end, &escaped,    &escaped_len))
      return 0;
    opt_flags= *buf++;
  }
  else
  {
    if (buf_end - buf < 7)
      return 0;                                 // Wrong data
    field_term_len= enclosed_len= line_term_len= line_start_len= escaped_len=1;
    field_term=  (char*) buf++;                 // Use first byte in string
    enclosed=    (char*) buf++;
    line_term=   (char*) buf++;
    line_start=  (char*) buf++;
    escaped=     (char*) buf++;
    opt_flags=   *buf++;
    empty_flags= *buf++;
    if (empty_flags & FIELD_TERM_EMPTY)
      field_term_len=0;
    if (empty_flags & ENCLOSED_EMPTY)
      enclosed_len=0;
    if (empty_flags & LINE_TERM_EMPTY)
      line_term_len=0;
    if (empty_flags & LINE_START_EMPTY)
      line_start_len=0;
    if (empty_flags & ESCAPED_EMPTY)
      escaped_len=0;
  }
  return buf;
}



/**************************************************************************
 Rows_log_event member functions
**************************************************************************/


Rows_log_event::Rows_log_event(const uchar *buf, size_t event_len,
                               const Format_description_log_event
                               *description_event)
  : Log_event(buf, description_event),
    m_row_count(0),
#ifndef MYSQL_CLIENT
    m_table(NULL),
#endif
    m_table_id(0), m_rows_buf(0), m_rows_cur(0), m_rows_end(0),
    m_extra_row_data(0)
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
    , m_curr_row(NULL), m_curr_row_end(NULL),
    m_key(NULL), m_key_info(NULL), m_key_nr(0),
    m_usable_key_parts(0), master_had_triggers(0)
#endif
{
  DBUG_ENTER("Rows_log_event::Rows_log_event(const char*,...)");
  uint8 const common_header_len= description_event->common_header_len;
  Log_event_type event_type= (Log_event_type)(uchar)buf[EVENT_TYPE_OFFSET];
  m_type= event_type;
  m_cols_ai.bitmap= 0; // Set to invalid, so it can be processed in is_valid().

  uint8 const post_header_len= description_event->post_header_len[event_type-1];

  if (event_len < (uint)(common_header_len + post_header_len))
    DBUG_VOID_RETURN;

  DBUG_PRINT("enter",("event_len: %zu  common_header_len: %d  "
        "post_header_len: %d",
        event_len, common_header_len,
        post_header_len));

  const uchar *post_start= buf + common_header_len;
  post_start+= RW_MAPID_OFFSET;
  if (post_header_len == 6)
  {
    /* Master is of an intermediate source tree before 5.1.4. Id is 4 bytes */
    m_table_id= uint4korr(post_start);
    post_start+= 4;
  }
  else
  {
    m_table_id= (ulonglong) uint6korr(post_start);
    post_start+= RW_FLAGS_OFFSET;
  }

  m_flags_pos= post_start - buf;
  m_flags= uint2korr(post_start);
  post_start+= 2;

  uint16 var_header_len= 0;
  if (post_header_len == ROWS_HEADER_LEN_V2)
  {
    /*
      Have variable length header, check length,
      which includes length bytes
    */
    var_header_len= uint2korr(post_start);
    /* Check length and also avoid out of buffer read */
    if (var_header_len < 2 ||
        event_len < static_cast<unsigned int>(var_header_len +
          (post_start - buf)))
    {
      m_cols.bitmap= 0;
      DBUG_VOID_RETURN;
    }
    var_header_len-= 2;

    /* Iterate over var-len header, extracting 'chunks' */
    const uchar *start= post_start + 2;
    const uchar *end= start + var_header_len;
    for (const uchar* pos= start; pos < end;)
    {
      switch(*pos++)
      {
      case RW_V_EXTRAINFO_TAG:
      {
        /* Have an 'extra info' section, read it in */
        if (unlikely((end - pos) <= EXTRA_ROW_INFO_LEN_OFFSET))
        {
          m_cols.bitmap= 0;
          DBUG_VOID_RETURN;
        }
        uint8 infoLen= pos[EXTRA_ROW_INFO_LEN_OFFSET];
        if (unlikely(infoLen < EXTRA_ROW_INFO_HDR_BYTES || (end-pos) < infoLen))
        {
          m_cols.bitmap= 0;
          DBUG_VOID_RETURN;
        }
        /* Just store/use the first tag of this type, skip others */
        if (likely(!m_extra_row_data))
        {
          m_extra_row_data= (uchar*) my_malloc(PSI_INSTRUMENT_ME, infoLen,
                                               MYF(MY_WME));
          if (likely(m_extra_row_data != NULL))
          {
            memcpy(m_extra_row_data, pos, infoLen);
          }
        }
        pos+= infoLen;
        break;
      }
      default:
        /* Unknown code, we will not understand anything further here */
        pos= end; /* Break loop */
      }
    }
  }

  uchar const *const var_start=
    (const uchar *)buf + common_header_len + post_header_len + var_header_len;
  uchar const *const ptr_width= var_start;
  uchar *ptr_after_width= (uchar*) ptr_width;
  DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));
  m_width= net_field_length(&ptr_after_width);
  DBUG_PRINT("debug", ("m_width=%u", m_width));

  /* Avoid reading out of buffer */
  if (ptr_after_width + (m_width + 7) / 8 > (uchar*)buf + event_len)
  {
    m_cols.bitmap= NULL;
    DBUG_VOID_RETURN;
  }

  /* if my_bitmap_init fails, caught in is_valid() */
  if (likely(!my_bitmap_init(&m_cols,
                             m_width <= sizeof(m_bitbuf)*8 ? m_bitbuf : NULL,
                             m_width)))
  {
    DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));
    bitmap_import(&m_cols, ptr_after_width);
    DBUG_DUMP("m_cols", (uchar*) ptr_after_width, no_bytes_in_export_map(&m_cols));
    ptr_after_width+= (m_width + 7) / 8;
  }
  else
    DBUG_VOID_RETURN;

  if (LOG_EVENT_IS_UPDATE_ROW(event_type))
  {
    DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));

    /* if my_bitmap_init fails, caught in is_valid() */
    if (likely(!my_bitmap_init(&m_cols_ai,
                               m_width <= sizeof(m_bitbuf_ai)*8 ? m_bitbuf_ai :
                               NULL,
                               m_width)))
    {
      DBUG_PRINT("debug", ("Reading from %p", ptr_after_width));
      bitmap_import(&m_cols_ai, ptr_after_width);
      DBUG_DUMP("m_cols_ai", ptr_after_width, no_bytes_in_export_map(&m_cols_ai));
      ptr_after_width+= (m_width + 7) / 8;
    }
    else
    {
      DBUG_ASSERT(m_cols_ai.bitmap == NULL);
      DBUG_VOID_RETURN;
    }
  }
  else
  {
    m_cols_ai= m_cols; /* Safety */
#ifdef DBUG_OFF
    /*
      m_cols_ai should only be usable for update events. Make sure nobody
      successfully manipulates it in debug builds.
    */
    m_cols_ai.bitmap= (my_bitmap_map*)1;
#endif
  }

  const uchar* const ptr_rows_data= (const uchar*) ptr_after_width;

  size_t const read_size= ptr_rows_data - (const unsigned char *) buf;
  if (read_size > event_len)
  {
    DBUG_VOID_RETURN;
  }
  size_t const data_size= event_len - read_size;
  DBUG_PRINT("info",("m_table_id: %llu  m_flags: %d  m_width: %u  data_size: %lu",
                     m_table_id, m_flags, m_width, (ulong) data_size));

  m_rows_buf= (uchar*) my_malloc(PSI_INSTRUMENT_ME, data_size, MYF(MY_WME));
  if (likely((bool)m_rows_buf))
  {
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
    m_curr_row= m_rows_buf;
#endif
    m_rows_end= m_rows_buf + data_size;
    m_rows_cur= m_rows_end;
    memcpy(m_rows_buf, ptr_rows_data, data_size);
    m_rows_before_size= ptr_rows_data - (const uchar *) buf; // Get the size that before SET part
  }
  else
    m_cols.bitmap= 0; // to not free it

  DBUG_VOID_RETURN;
}

void Rows_log_event::uncompress_buf()
{
  uint32 un_len= binlog_get_uncompress_len(m_rows_buf);
  if (unlikely(!un_len || un_len > MAX_MAX_ALLOWED_PACKET))
  {
    /* my_bitmap_free() nulls m_cols.bitmap, which is_valid() rejects. */
    my_bitmap_free(&m_cols);
    return;
  }

  uchar *new_buf= (uchar*) my_malloc(PSI_INSTRUMENT_ME, ALIGN_SIZE(un_len),
                                     MYF(MY_WME));
  if (new_buf)
  {
    if (!binlog_buf_uncompress(m_rows_buf, new_buf,
                              (uint32)(m_rows_cur - m_rows_buf), &un_len))
    {
      my_free(m_rows_buf);
      m_rows_buf= new_buf;
#if !defined(MYSQL_CLIENT) && defined(HAVE_REPLICATION)
      m_curr_row= m_rows_buf;
#endif
      m_rows_end= m_rows_buf + un_len;
      m_rows_cur= m_rows_end;
      return;
    }
    else
    {
      my_free(new_buf);
    }
  }
  my_bitmap_free(&m_cols); // catch it in is_valid
}

Rows_log_event::~Rows_log_event()
{
  my_bitmap_free(&m_cols); // To pair with my_bitmap_init().
  my_free(m_rows_buf);
  my_free(m_extra_row_data);
}

int Rows_log_event::get_data_size()
{
  int const general_type_code= get_general_type_code();

  uchar buf[MAX_INT_WIDTH];
  uchar *end= net_store_length(buf, m_width);

  DBUG_EXECUTE_IF("old_row_based_repl_4_byte_map_id_master",
                  return (int) (6 + no_bytes_in_export_map(&m_cols) + (end - buf) +
                                (general_type_code == UPDATE_ROWS_EVENT ?
                                 no_bytes_in_export_map(&m_cols_ai) : 0) +
                                m_rows_cur - m_rows_buf););
  int data_size= 0;
  Log_event_type type= get_type_code();
  bool is_v2_event= LOG_EVENT_IS_ROW_V2(type);
  if (is_v2_event)
  {
    data_size= ROWS_HEADER_LEN_V2 +
      (m_extra_row_data ?
       RW_V_TAG_LEN + m_extra_row_data[EXTRA_ROW_INFO_LEN_OFFSET]:
       0);
  }
  else
  {
    data_size= ROWS_HEADER_LEN_V1;
  }
  data_size+= no_bytes_in_export_map(&m_cols);
  data_size+= (uint) (end - buf);

  if (general_type_code == UPDATE_ROWS_EVENT)
    data_size+= no_bytes_in_export_map(&m_cols_ai);

  data_size+= (uint) (m_rows_cur - m_rows_buf);
  return data_size; 
}


/**************************************************************************
 Annotate_rows_log_event member functions
**************************************************************************/

Annotate_rows_log_event::
Annotate_rows_log_event(const uchar *buf,
                        uint event_len,
                        const Format_description_log_event *desc)
  : Log_event(buf, desc),
    m_save_thd_query_txt(0),
    m_save_thd_query_len(0),
    m_saved_thd_query(false),
    m_used_query_txt(0)
{
  m_query_len= event_len - desc->common_header_len;
  m_query_txt= (char*) buf + desc->common_header_len;
}

Annotate_rows_log_event::~Annotate_rows_log_event()
{
  DBUG_ENTER("Annotate_rows_log_event::~Annotate_rows_log_event");
#ifndef MYSQL_CLIENT
  if (m_saved_thd_query)
    thd->set_query(m_save_thd_query_txt, m_save_thd_query_len);
  else if (m_used_query_txt)
    thd->reset_query();
#endif
  DBUG_VOID_RETURN;
}

int Annotate_rows_log_event::get_data_size()
{
  return m_query_len;
}

Log_event_type Annotate_rows_log_event::get_type_code()
{
  return ANNOTATE_ROWS_EVENT;
}

bool Annotate_rows_log_event::is_valid() const
{
  return (m_query_txt != NULL && m_query_len != 0);
}


/**************************************************************************
 Table_map_log_event member functions and support functions
**************************************************************************/

/**
  @page How replication of field metadata works.
  
  When a table map is created, the master first calls 
  Table_map_log_event::save_field_metadata() which calculates how many 
  values will be in the field metadata. Only those fields that require the 
  extra data are added. The method also loops through all of the fields in 
  the table calling the method Field::save_field_metadata() which returns the
  values for the field that will be saved in the metadata and replicated to
  the slave. Once all fields have been processed, the table map is written to
  the binlog adding the size of the field metadata and the field metadata to
  the end of the body of the table map.

  When a table map is read on the slave, the field metadata is read from the 
  table map and passed to the table_def class constructor which saves the 
  field metadata from the table map into an array based on the type of the 
  field. Field metadata values not present (those fields that do not use extra 
  data) in the table map are initialized as zero (0). The array size is the 
  same as the columns for the table on the slave.

  Additionally, values saved for field metadata on the master are saved as a 
  string of bytes (uchar) in the binlog. A field may require 1 or more bytes
  to store the information. In cases where values require multiple bytes 
  (e.g. values > 255), the endian-safe methods are used to properly encode 
  the values on the master and decode them on the slave. When the field
  metadata values are captured on the slave, they are stored in an array of
  type uint16. This allows the least number of casts to prevent casting bugs
  when the field metadata is used in comparisons of field attributes. When
  the field metadata is used for calculating addresses in pointer math, the
  type used is uint32. 
*/

/*
  Constructor used by slave to read the event from the binary log.
 */
#if defined(HAVE_REPLICATION)
Table_map_log_event::Table_map_log_event(const uchar *buf, uint event_len,
                                         const Format_description_log_event
                                         *description_event)

  : Log_event(buf, description_event),
#ifndef MYSQL_CLIENT
    m_table(NULL),
#endif
    m_dbnam(NULL), m_dblen(0), m_tblnam(NULL), m_tbllen(0),
    m_colcnt(0), m_coltype(0),
    m_memory(NULL), m_table_id(ULONGLONG_MAX), m_flags(0),
    m_data_size(0), m_field_metadata(0), m_field_metadata_size(0),
    m_null_bits(0), m_meta_memory(NULL),
    m_optional_metadata_len(0), m_optional_metadata(NULL)
{
  unsigned int bytes_read= 0;
  DBUG_ENTER("Table_map_log_event::Table_map_log_event(const char*,uint,...)");

  uint8 common_header_len= description_event->common_header_len;
  uint8 post_header_len= description_event->post_header_len[TABLE_MAP_EVENT-1];
  DBUG_PRINT("info",("event_len: %u  common_header_len: %d  post_header_len: %d",
                     event_len, common_header_len, post_header_len));

  /*
    Don't print debug messages when running valgrind since they can
    trigger false warnings.
   */
#ifndef HAVE_valgrind
  DBUG_DUMP("event buffer", (uchar*) buf, event_len);
#endif

 if (event_len < (uint)(common_header_len + post_header_len))
  DBUG_VOID_RETURN;

  /* Read the post-header */
  const uchar *post_start= buf + common_header_len;

  post_start+= TM_MAPID_OFFSET;
  VALIDATE_BYTES_READ(post_start, buf, event_len);
  if (post_header_len == 6)
  {
    /* Master is of an intermediate source tree before 5.1.4. Id is 4 bytes */
    m_table_id= uint4korr(post_start);
    post_start+= 4;
  }
  else
  {
    DBUG_ASSERT(post_header_len == TABLE_MAP_HEADER_LEN);
    m_table_id= (ulonglong) uint6korr(post_start);
    post_start+= TM_FLAGS_OFFSET;
  }

  DBUG_ASSERT((m_table_id & MAX_TABLE_MAP_ID) != UINT32_MAX &&
              (m_table_id & MAX_TABLE_MAP_ID) != 0);

  m_flags= uint2korr(post_start);

  /* Read the variable part of the event */
  const uchar *const vpart= buf + common_header_len + post_header_len;

  /* Extract the length of the various parts from the buffer */
  uchar const *const ptr_dblen= (uchar const*)vpart + 0;
  VALIDATE_BYTES_READ(ptr_dblen, buf, event_len);
  m_dblen= *(uchar*) ptr_dblen;

  /* Length of database name + counter + terminating null */
  uchar const *const ptr_tbllen= ptr_dblen + m_dblen + 2;
  VALIDATE_BYTES_READ(ptr_tbllen, buf, event_len);
  m_tbllen= *(uchar*) ptr_tbllen;

  /*
    If either the database or table name is too long, return before allocating
    memory for the event, so future validity checks on this event (i.e.
    is_valid()) return false.
  */
  if (unlikely(m_dblen > NAME_LEN || m_tbllen > NAME_LEN))
    DBUG_VOID_RETURN;

  /* Length of table name + counter + terminating null */
  uchar const *const ptr_colcnt= ptr_tbllen + m_tbllen + 2;
  uchar *ptr_after_colcnt= (uchar*) ptr_colcnt;
  VALIDATE_BYTES_READ(ptr_after_colcnt, buf, event_len);
  m_colcnt= net_field_length(&ptr_after_colcnt);
  DBUG_EXECUTE_IF("corrupt_table_map_colcnt_read",
                  m_colcnt= (1 << 20););

  DBUG_PRINT("info",("m_dblen: %lu  off: %ld  m_tbllen: %lu  off: %ld  m_colcnt: %lu  off: %ld",
                     (ulong) m_dblen, (long) (ptr_dblen - vpart),
                     (ulong) m_tbllen, (long) (ptr_tbllen - vpart),
                     m_colcnt, (long) (ptr_colcnt - vpart)));

  /* Allocate mem for all fields in one go. If fails, caught in is_valid() */
  m_memory= (uchar*) my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME),
                                     &m_dbnam, (uint) m_dblen + 1,
                                     &m_tblnam, (uint) m_tbllen + 1,
                                     &m_coltype, (uint) m_colcnt,
                                     NullS);

  if (m_memory)
  {
    /* Copy the different parts into their memory */
    strncpy(const_cast<char*>(m_dbnam), (const char*)ptr_dblen  + 1, m_dblen + 1);
    strncpy(const_cast<char*>(m_tblnam), (const char*)ptr_tbllen + 1, m_tbllen + 1);
    /*
      Future uses of the database and table name require a null-terminating
      string, but the source string from the above copy may not have one, and
      strncpy will not add one. So always add it to ensure it exists.
    */
    const_cast<char*>(m_dbnam)[m_dblen]= '\0';
    const_cast<char*>(m_tblnam)[m_tbllen]= '\0';
    if (unlikely(ptr_after_colcnt + m_colcnt > buf + event_len))
    {
      my_free(m_memory);
      m_memory= NULL;
      DBUG_VOID_RETURN;
    }
    memcpy(m_coltype, ptr_after_colcnt, m_colcnt);

    ptr_after_colcnt= ptr_after_colcnt + m_colcnt;
    VALIDATE_BYTES_READ(ptr_after_colcnt, buf, event_len);
    m_field_metadata_size= net_field_length(&ptr_after_colcnt);
    DBUG_EXECUTE_IF("corrupt_table_map_field_metadata_size_read",
                    m_field_metadata_size= (1 << 20););

    /*
      Ensure the field metadata block covers the metadata the declared
      column types consume.
    */
    size_t metadata_needed= 0;
    for (ulong i= 0; i < m_colcnt; i++)
      metadata_needed+= table_def::field_metadata_length(m_coltype[i]);
    if (unlikely(m_field_metadata_size < metadata_needed))
    {
      m_coltype= NULL;
      my_free(m_memory);
      m_memory= NULL;
      DBUG_VOID_RETURN;
    }

    if (m_field_metadata_size <= (m_colcnt * 2))
    {
      uint num_null_bytes= (m_colcnt + 7) / 8;
      m_meta_memory= (uchar *)my_multi_malloc(PSI_INSTRUMENT_ME, MYF(MY_WME),
          &m_null_bits, num_null_bytes,
          &m_field_metadata, m_field_metadata_size,
          NULL);
      if (unlikely(ptr_after_colcnt + m_field_metadata_size > buf + event_len))
      {
        my_free(m_meta_memory);
        m_meta_memory= NULL;
        my_free(m_memory);
        m_memory= NULL;
        DBUG_VOID_RETURN;
      }
      memcpy(m_field_metadata, ptr_after_colcnt, m_field_metadata_size);
      ptr_after_colcnt= (uchar*)ptr_after_colcnt + m_field_metadata_size;
      if (unlikely(ptr_after_colcnt + num_null_bytes > buf + event_len))
      {
        my_free(m_meta_memory);
        m_meta_memory= NULL;
        my_free(m_memory);
        m_memory= NULL;
        DBUG_VOID_RETURN;
      }
      memcpy(m_null_bits, ptr_after_colcnt, num_null_bytes);
      ptr_after_colcnt= (unsigned char*)ptr_after_colcnt + num_null_bytes;
    }
    else
    {
      m_coltype= NULL;
      my_free(m_memory);
      m_memory= NULL;
      DBUG_VOID_RETURN;
    }

    bytes_read= (uint) (ptr_after_colcnt - (uchar *)buf);

    /* After null_bits field, there are some new fields for extra metadata. */
    if (bytes_read < event_len)
    {
      m_optional_metadata_len= event_len - bytes_read;
      m_optional_metadata=
        static_cast<unsigned char*>(my_malloc(PSI_INSTRUMENT_ME, m_optional_metadata_len, MYF(MY_WME)));
      memcpy(m_optional_metadata, ptr_after_colcnt, m_optional_metadata_len);
    }
  }
#ifdef MYSQL_SERVER
  if (!m_table)
    DBUG_VOID_RETURN;
  binlog_type_info_array= thd->alloc<Binlog_type_info>(m_table->s->fields);
  for (uint i= 0; i <  m_table->s->fields; i++)
    binlog_type_info_array[i]= m_table->field[i]->binlog_type_info();
#endif

  DBUG_VOID_RETURN;
}
#endif

Table_map_log_event::~Table_map_log_event()
{
  my_free(m_meta_memory);
  my_free(m_memory);
  my_free(m_optional_metadata);
  m_optional_metadata= NULL;
}

/**
   Parses SIGNEDNESS field.

   @param[out] column_metadata store all column metadata here (including signedness)
   @param[in]  field   SIGNEDNESS field in table_map_event.
   @param[in]  length  length of the field
 */
using Optional_column_metadata= Table_map_log_event::Optional_metadata_fields::Column_metadata;
static void parse_signedness(
  Dynamic_array<Optional_column_metadata>& column_metadata,
  unsigned char *field, unsigned int length)
{
  unsigned char *p= field;
  unsigned char *field_end= p + length;
  unsigned char unsigned_bitfield= 0;
  unsigned char mask= 0;

  for (uint col= 0; col < column_metadata.size(); col++)
  {
    auto &col_meta= column_metadata.at(col);
    if (!is_numeric_type(col_meta.column_type))
      continue;
    if (mask == 0)
    {
      if (p >= field_end)
        return;
      unsigned_bitfield= *p++;
      mask= 0x80;
    }
    col_meta.is_unsigned= unsigned_bitfield & mask;
    mask >>= 1;
  }
}

/**
   Parses DEFAULT_CHARSET field.

   @param[out] fields  store all optional metadata here (including charsets)
   @param[in]  field   DEFAULT_CHARSET field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_default_charset(
  Table_map_log_event::Optional_metadata_fields& fields,
  unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  uint default_cs= net_field_length(&p);
  if (unlikely(p > end))
    return;
  for (uint i= 0; i < fields.m_column_metadata.size(); i++)
    if (is_character_type(fields.m_column_metadata.at(i).column_type))
      fields.m_column_metadata.at(i).charset= default_cs;
  while (p < end)
  {
    unsigned int col_index= net_field_length(&p);
    if (unlikely(p > end))
      return;
    uint charset= net_field_length(&p);
    if (unlikely(p > end))
      return;
    if (col_index < fields.m_column_metadata.size())
      fields.m_column_metadata.at(col_index).charset= charset;
  }
}

/**
   Parses ENUM_AND_SET_DEFAULT_CHARSET field.

   @param[out] fields  store all optional metadata here (including charsets)
   @param[in]  field   ENUM_AND_SET_DEFAULT_CHARSET field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_enum_and_set_default_charset(
  Table_map_log_event::Optional_metadata_fields& fields,
  unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  uint default_cs= net_field_length(&p);
  if (unlikely(p > end))
    return;
  for (uint i= 0; i < fields.m_column_metadata.size(); i++)
    if (is_enum_or_set_type(fields.m_column_metadata.at(i).column_type))
      fields.m_column_metadata.at(i).enum_and_set_column_charset= default_cs;
  while (p < end)
  {
    unsigned int col_index= net_field_length(&p);
    if (unlikely(p > end))
      return;
    uint charset= net_field_length(&p);
    if (unlikely(p > end))
      return;
    if (col_index < fields.m_column_metadata.size())
      fields.m_column_metadata.at(col_index).enum_and_set_column_charset= charset;
  }
}

/**
   Parses COLUMN_CHARSET field.

   @param[out] column_metadata store all column metadata here (including charsets)
   @param[in]  field   COLUMN_CHARSET field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_column_charset(
  Dynamic_array<Optional_column_metadata>& column_metadata,
  unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  for (uint col= 0; p < end && col < column_metadata.size(); col++)
  {
    if (!is_character_type(column_metadata.at(col).column_type))
      continue;
    column_metadata.at(col).charset= net_field_length(&p);
    if (unlikely(p > end))
      return;
  }
}

/**
   Parses ENUM_AND_SET_COLUMN_CHARSET field.

   @param[out] column_metadata store all column metadata here (including charsets)
   @param[in]  field   ENUM_AND_SET_COLUMN_CHARSET field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_enum_and_set_column_charset(
  Dynamic_array<Optional_column_metadata>& column_metadata,
  unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  for (uint col= 0; p < end && col < column_metadata.size(); col++)
  {
    if (!is_enum_or_set_type(column_metadata.at(col).column_type))
      continue;
    column_metadata.at(col).enum_and_set_column_charset= net_field_length(&p);
    if (unlikely(p > end))
      return;
  }
}

/**
   Parses COLUMN_NAME field.

   @param[in]  root    Allocate memory here
   @param[out] column_metadata store all column metadata here (including column names)
   @param[in]  field   COLUMN_NAME field in table_map_event.
   @param[in]  length  length of the field
 */
static bool parse_column_name(MEM_ROOT *root,
  Dynamic_array<Optional_column_metadata>& column_metadata,
  unsigned char *field, unsigned int length)
{
  unsigned int col = 0;
  for (uchar *end= field+length; field < end && col < column_metadata.size(); col++)
  {
    LEX_CSTRING& name = column_metadata.at(col).column_name;
    uint name_length= net_field_length(&field);
    if (unlikely(field + name_length > end))
      return 1;
    if (!(name.str= strmake_root(root, (char*) field, name_length)))
      return 1;
    name.length= name_length;
    field+= name_length;
  }

  return 0;
}

/**
   Parses SET_STR_VALUE/ENUM_STR_VALUE field.

   @param[out] column_metadata store all column metadata here (including
               set/enum strings)
   @param[in]  field      SET_STR_VALUE/ENUM_STR_VALUE field in table_map_event.
   @param[in]  length     length of the field
   @param[in]  column_type  type of the given column (i.e. MYSQL_TYPE_XYZ)
   @param[in]  var_to_set   pointer-to-member selecting which str_vector to fill

   @retval false ok
   @retval true  out of memory while reserving the str_vector
 */
static bool parse_set_str_value(
    Dynamic_array<Optional_column_metadata> &column_metadata,
    unsigned char *field, unsigned int length,
    uchar column_type,
    Table_map_log_event::Optional_metadata_fields::str_vector
        Optional_column_metadata::*var_to_set)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  for (uint col= 0; p < end && col < column_metadata.size(); col++)
  {
    if (column_metadata.at(col).column_type != column_type)
      continue;

    unsigned int count= net_field_length(&p);
    if (unlikely(p > end))
      return 0;

    auto* column_strings= &(column_metadata.at(col).*var_to_set);

    if (column_strings->reserve(count))
      return 1;

    for (unsigned int i= 0; i < count; i++)
    {
      unsigned len1= net_field_length(&p);
      if (unlikely(p + len1 > end))
        return 0;
      column_strings->append(LEX_CSTRING{reinterpret_cast<char *>(p), len1});
      p+= len1;
    }
  }

  return 0;
}

/**
   Parses GEOMETRY_TYPE field.

   @param[out] column_metadata store all column metadata here (including
     geometry types)
   @param[in]  field   GEOMETRY_TYPE field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_geometry_type(
    Dynamic_array<Optional_column_metadata> &column_metadata,
    unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  for (unsigned int col = 0; p < end && col < column_metadata.size(); col++)
  {
    auto& col_metadata = column_metadata.at(col);
    if (col_metadata.column_type == MYSQL_TYPE_GEOMETRY)
    {
      col_metadata.geometry_type = net_field_length(&p);
      if (unlikely(p > end))
        return;
    }
  }
}

/**
   Parses SIMPLE_PRIMARY_KEY field.

   @param[out] column_metadata store all column metadata here (including
     primary key information). The prefix is always 0 for SIMPLE_PRIMARY_KEY.
   @param[in]  field   SIMPLE_PRIMARY_KEY field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_simple_pk(
    Dynamic_array<Optional_column_metadata> &column_metadata,
    unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  while (p < end)
  {
    unsigned int col_index= net_field_length(&p);
    if (unlikely(p > end))
      return;
    if (col_index < column_metadata.size())
      column_metadata.at(col_index).primary_key = 0;
  }
}

/**
   Parses PRIMARY_KEY_WITH_PREFIX field.

   @param[out] column_metadata store all column metadata here (including
     primary key lengths)
   @param[in]  field   PRIMARY_KEY_WITH_PREFIX field in table_map_event.
   @param[in]  length  length of the field
 */
static void parse_pk_with_prefix(
    Dynamic_array<Optional_column_metadata> &column_metadata,
    unsigned char *field, unsigned int length)
{
  unsigned char* p= field;
  unsigned char* end= field + length;

  while (p < end)
  {
    unsigned int col_index= net_field_length(&p);
    if (unlikely(p > end))
      return;
    unsigned int col_prefix= net_field_length(&p);
    if (unlikely(p > end))
      return;
    if (col_index < column_metadata.size())
      column_metadata.at(col_index).primary_key = col_prefix;
  }
}

Table_map_log_event::Optional_metadata_fields::
Optional_metadata_fields(MEM_ROOT *root, uint master_columns,
                         const uchar* column_types,
                         const uchar* field_metadata,
                         uchar* optional_metadata,
                         size_t optional_metadata_len,
                         bool only_column_names)
    : m_column_metadata(PSI_INSTRUMENT_MEM, master_columns)
{
  unsigned int len;
  uchar *metadata_end;

  allocation_error= 0;

  // initializing the elements of m_column_metadata and their types
  for (uint i= 0; i < master_columns; i++)
  {
    Column_metadata col{};
    col.column_type = column_types[i];
    if (field_metadata)
    {
      if (col.column_type == MYSQL_TYPE_STRING &&
          (*field_metadata == MYSQL_TYPE_ENUM ||
           *field_metadata == MYSQL_TYPE_SET))
        col.column_type= *field_metadata;
      advance_field_metadata_ptr(col.column_type, &field_metadata);
    }
    if (m_column_metadata.append(col))
      goto error;
  }

  if (optional_metadata == NULL)
    return;

  metadata_end= optional_metadata + optional_metadata_len;
  for (uchar *field= optional_metadata ; field < metadata_end ; field+= len)
  {
    Optional_metadata_field_type type=
      static_cast<Optional_metadata_field_type>(field[0]);

    // Get length and move field to the value.
    field++;
    len= net_field_length(&field);
    if (only_column_names && type != COLUMN_NAME)
      continue;

    switch(type) {
    case SIGNEDNESS:
      parse_signedness(m_column_metadata, field, len);
      break;
    case DEFAULT_CHARSET:
      parse_default_charset(*this, field, len);
      break;
    case COLUMN_CHARSET:
      parse_column_charset(m_column_metadata, field, len);
      break;
    case COLUMN_NAME:
      if (parse_column_name(root, m_column_metadata, field, len))
        goto error;
      break;
    case SET_STR_VALUE:
      if (parse_set_str_value(m_column_metadata, field, len, MYSQL_TYPE_SET,
                              &Optional_column_metadata::set_str_values))
        goto error;
      break;
    case ENUM_STR_VALUE:
      if (parse_set_str_value(m_column_metadata, field, len, MYSQL_TYPE_ENUM,
                              &Optional_column_metadata::enum_str_values))
        goto error;
      break;
    case GEOMETRY_TYPE:
      parse_geometry_type(m_column_metadata, field, len);
      break;
    case SIMPLE_PRIMARY_KEY:
      parse_simple_pk(m_column_metadata, field, len);
      break;
    case PRIMARY_KEY_WITH_PREFIX:
      parse_pk_with_prefix(m_column_metadata, field, len);
      break;
    case ENUM_AND_SET_DEFAULT_CHARSET:
      parse_enum_and_set_default_charset(*this, field, len);
      break;
    case ENUM_AND_SET_COLUMN_CHARSET:
      parse_enum_and_set_column_charset(m_column_metadata, field, len);
      break;
    default:
      DBUG_ASSERT(0);
    }
  }
  return;

error:
  allocation_error= 1;
  /*
    The column names may be partially filled if parsing failed midway. Clear
    all of them so callers checking the first column's name fall back to
    positional column mapping instead of using a half-parsed name list.
  */
  for (size_t i= 0; i < m_column_metadata.size(); i++)
    m_column_metadata.at(i).column_name= LEX_CSTRING{nullptr, 0};
  return;
}


/**************************************************************************
 Write_rows_log_event member functions
**************************************************************************/


/*
  Constructor used by slave to read the event from the binary log.
 */
#ifdef HAVE_REPLICATION
Write_rows_log_event::Write_rows_log_event(const uchar *buf, size_t event_len,
                                           const Format_description_log_event
                                           *description_event)
: Rows_log_event(buf, event_len, description_event)
{
}

Write_rows_compressed_log_event::Write_rows_compressed_log_event(
                                           const uchar *buf, size_t event_len,
                                           const Format_description_log_event
                                           *description_event)
: Write_rows_log_event(buf, event_len, description_event)
{
  uncompress_buf();
}
#endif


/**************************************************************************
 Delete_rows_log_event member functions
**************************************************************************/

/*
  Constructor used by slave to read the event from the binary log.
 */
#ifdef HAVE_REPLICATION
Delete_rows_log_event::Delete_rows_log_event(const uchar *buf, size_t event_len,
                                             const Format_description_log_event
                                             *description_event)
  : Rows_log_event(buf, event_len, description_event)
{
}

Delete_rows_compressed_log_event::Delete_rows_compressed_log_event(
                                           const uchar *buf, size_t event_len,
                                           const Format_description_log_event
                                           *description_event)
  : Delete_rows_log_event(buf, event_len, description_event)
{
  uncompress_buf();
}
#endif

/**************************************************************************
 Update_rows_log_event member functions
**************************************************************************/

Update_rows_log_event::~Update_rows_log_event()
{
  my_bitmap_free(&m_cols_ai); // To pair with my_bitmap_init().
}


/*
  Constructor used by slave to read the event from the binary log.
 */
#ifdef HAVE_REPLICATION
Update_rows_log_event::Update_rows_log_event(
    const uchar *buf, size_t event_len,
    const Format_description_log_event *description_event)
    : Rows_log_event(buf, event_len, description_event)
{
}

Update_rows_compressed_log_event::Update_rows_compressed_log_event(
    const uchar *buf, size_t event_len,
    const Format_description_log_event *description_event)
    : Update_rows_log_event(buf, event_len, description_event)
{
  uncompress_buf();
}
#endif

Incident_log_event::Incident_log_event(const uchar *buf, uint event_len,
                                       const Format_description_log_event *descr_event)
  : Log_event(buf, descr_event)
{
  DBUG_ENTER("Incident_log_event::Incident_log_event");
  uint8 const common_header_len=
    descr_event->common_header_len;
  uint8 const post_header_len=
    descr_event->post_header_len[INCIDENT_EVENT-1];

  DBUG_PRINT("info",("event_len: %u; common_header_len: %d; post_header_len: %d",
                     event_len, common_header_len, post_header_len));

  m_message.str= NULL;
  m_message.length= 0;
  int incident_number= uint2korr(buf + common_header_len);
  if (incident_number >= INCIDENT_COUNT ||
      incident_number <= INCIDENT_NONE)
  {
    // If the incident is not recognized, this binlog event is
    // invalid.  If we set incident_number to INCIDENT_NONE, the
    // invalidity will be detected by is_valid().
    m_incident= INCIDENT_NONE;
    DBUG_VOID_RETURN;
  }
  m_incident= static_cast<Incident>(incident_number);
  uchar const *ptr= buf + common_header_len + post_header_len;
  uchar const *const str_end= buf + event_len;
  uint8 len= 0;                   // Assignment to keep compiler happy
  const char *str= NULL;          // Assignment to keep compiler happy
  if (read_str(&ptr, str_end, &str, &len))
  {
    /* Mark this event invalid */
    m_incident= INCIDENT_NONE;
    DBUG_VOID_RETURN;
  }
  if (!(m_message.str= (char*) my_malloc(key_memory_log_event, len+1, MYF(MY_WME))))
  {
    /* Mark this event invalid */
    m_incident= INCIDENT_NONE;
    DBUG_VOID_RETURN;
  }
  strmake(m_message.str, str, len);
  m_message.length= len;
  DBUG_PRINT("info", ("m_incident: %d", m_incident));
  DBUG_VOID_RETURN;
}


Partial_rows_log_event::Partial_rows_log_event(
    const uchar *buf, uint event_len,
    const Format_description_log_event *description_event)
    : Log_event(buf, description_event), metadata_written(0), rows_event(NULL)
{
  DBUG_ENTER("Partial_rows_log_event::Partial_rows_log_even(const uchar*,uint,...)");

  uint8 common_header_len= description_event->common_header_len;
  uint8 post_header_len= description_event->post_header_len[PARTIAL_ROW_DATA_EVENT-1];
  DBUG_PRINT("info",("event_len: %u  common_header_len: %d  post_header_len: %d",
                     event_len, common_header_len, post_header_len));
  DBUG_ASSERT(post_header_len == PARTIAL_ROWS_HEADER_LEN);

 if (event_len < (uint)(common_header_len + post_header_len))
  DBUG_VOID_RETURN;

  start_offset= common_header_len + PARTIAL_ROWS_HEADER_LEN;
  ev_buffer_base= buf;
  end_offset= event_len;

  /* Read the post-header */
  const uchar *post_start= buf + common_header_len;
  VALIDATE_BYTES_READ(post_start, buf, event_len);

  total_fragments= uint4korr(post_start);
  post_start+= 4;
  VALIDATE_BYTES_READ(post_start, buf, event_len);

  seq_no= uint4korr((post_start));
  post_start+= 4;
  VALIDATE_BYTES_READ(post_start, buf, event_len);
  DBUG_ASSERT(seq_no <= total_fragments);

  flags2= *(post_start++);
  VALIDATE_BYTES_READ(post_start, buf, event_len);

  if (flags2 & FL_ORIG_EVENT_SIZE)
  {
    DBUG_ASSERT(seq_no == 1);
    original_event_size= uint8korr((post_start));
    post_start+= 8;
    start_offset+= 8;
    VALIDATE_BYTES_READ(post_start, buf, event_len);
    DBUG_ASSERT(original_event_size);
  }

  /*
    post_start so far only has read header data, no data from the rows event
    itself. Ensure there is still data for the rows event.
  */
  DBUG_ASSERT(static_cast<uint>(post_start - ev_buffer_base) < event_len);

  DBUG_VOID_RETURN;
}

bool Partial_rows_log_event::is_valid() const {
  bool is_first= seq_no == 1;
  /*
    There should be at least 2 fragments for the group, and the sequence
    number of this instance should fall between 1 and the total number of
    fragments.
  */
  bool has_valid_sequencing=
      seq_no >= 1 && total_fragments >= 2 && seq_no <= total_fragments;

  bool has_valid_orig_event_size=
      ((flags2 & FL_ORIG_EVENT_SIZE) && is_first && original_event_size) ||
      !(flags2 & FL_ORIG_EVENT_SIZE);

  /*
    To be valid from a fragmentation context, the following should hold true:
      1) There should be a _valid_ Rows_log_event that is referenced
      2) Start_offset and end_offset should be representative of the
         fragment. If it is the first fragment, then start_offset should be
         0. The end_offset should be after the start_offset.
      3) If it is the first event, we should write metadata and the original
         event size; otherwise, neither should be written.
      4) ev_buffer_base should not be used
  */
  bool is_valid_for_fragmentation=
      (rows_event && rows_event->is_valid()) &&
      (((start_offset || is_first) && end_offset) &&
       (end_offset > start_offset)) &&
      ((is_first && metadata_written) || !metadata_written) &&
      !ev_buffer_base;

  /*
    To be valid from an assembly context, the following should hold true:
      1) ev_buffer_base should be set (referencing the read-in buffer)
      2) start_offset should point beyond the Partial_rows_log_event header,
         as it points to the start of the Rows_log_event content
      3) end_offset should be non-zero (there should always be some content)
      4) Neither rows_event nor metadata_written should be set
  */
  bool is_valid_for_assembly=
      ev_buffer_base && start_offset > PARTIAL_ROWS_HEADER_LEN &&
      end_offset && (!rows_event && !metadata_written);

  bool is_valid= has_valid_sequencing && has_valid_orig_event_size &&
                 (is_valid_for_fragmentation ^ is_valid_for_assembly);

  return is_valid;
}


Incident_log_event::~Incident_log_event()
{
  if (m_message.str)
    my_free(m_message.str);
}


const char *
Incident_log_event::description() const
{
  static const char *const description[]= {
    "NOTHING",                                  // Not used
    "LOST_EVENTS"
  };

  DBUG_PRINT("info", ("m_incident: %d", m_incident));
  return description[m_incident];
}


Ignorable_log_event::Ignorable_log_event(const uchar *buf,
                                         const Format_description_log_event
                                         *descr_event,
                                         const char *event_name)
  :Log_event(buf, descr_event), number((int) (uchar) buf[EVENT_TYPE_OFFSET]),
   description(event_name)
{
  DBUG_ENTER("Ignorable_log_event::Ignorable_log_event");
  DBUG_VOID_RETURN;
}

Ignorable_log_event::~Ignorable_log_event() = default;

bool copy_event_cache_to_file_and_reinit(IO_CACHE *cache, FILE *file)
{
  return (my_b_copy_all_to_file(cache, file) ||
          reinit_io_cache(cache, WRITE_CACHE, 0, FALSE, TRUE));
}

#if defined(MYSQL_SERVER) && defined(HAVE_REPLICATION)
int Log_event::apply_event(rpl_group_info* rgi)
{
  int res;
  THD_STAGE_INFO(thd, stage_apply_event);
  rgi->current_event= this;
  res= do_apply_event(rgi);
  rgi->current_event= NULL;
  THD_STAGE_INFO(thd, stage_after_apply_event);
  return res;
}
#endif