mysqlbinlog.cc

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

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

   This program is also distributed with certain software (including
   but not limited to OpenSSL) that is licensed under separate terms,
   as designated in a particular file or component or in included license
   documentation.  The authors of MySQL hereby grant you an additional
   permission to link the program and your derivative works with the
   separately licensed software that they have included with MySQL.

   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, version 2.0, 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-1301  USA
*/

/*
   Standalone program to read a MySQL binary log (or relay log).

   Should be able to read any file of these categories, even with
   --start-position.
   An important fact: the Format_desc event of the log is at most the 3rd event
   of the log; if it is the 3rd then there is this combination:
   Format_desc_of_slave, Rotate_of_master, Format_desc_of_master.
*/

#include "client/mysqlbinlog.h"

#include <fcntl.h>
#include <inttypes.h>
#include <signal.h>
#include <stdarg.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <algorithm>
#include <map>
#include <string>
#include <utility>

#include "caching_sha2_passwordopt-vars.h"
#include "client/client_priv.h"
#include "compression.h"
#include "libbinlogevents/include/codecs/factory.h"
#include "libbinlogevents/include/compression/factory.h"
#include "libbinlogevents/include/compression/iterator.h"
#include "libbinlogevents/include/trx_boundary_parser.h"
#include "my_byteorder.h"
#include "my_dbug.h"
#include "my_default.h"
#include "my_dir.h"
#include "my_io.h"
#include "my_macros.h"
#include "my_time.h"
#include "mysqld_error.h"
#include "prealloced_array.h"
#include "print_version.h"
#include "sql/binlog_reader.h"
#include "sql/log_event.h"
#include "sql/my_decimal.h"
#include "sql/rpl_constants.h"
#include "sql/rpl_dbtid.h"
#include "sql/rpl_gtid.h"
#include "sql_common.h"
#include "sql_string.h"
#include "sslopt-vars.h"
#include "typelib.h"
#include "welcome_copyright_notice.h"  // ORACLE_WELCOME_COPYRIGHT_NOTICE

#include <tuple>

using std::map;
using std::max;
using std::min;
using std::string;

/**
  For storing information of the Format_description_event of the currently
  active binlog. it will be changed each time a new Format_description_event is
  found in the binlog.
*/
Format_description_event glob_description_event(BINLOG_VERSION, server_version);

/**
  This class abstracts the rewriting of databases for RBR events.
 */
class Database_rewrite {
 public:
  using Rewrite_result =
      std::tuple<unsigned char *, std::size_t, std::size_t, bool>;

 private:
  class Transaction_payload_content_rewriter {
    using Rewrite_payload_result = std::tuple<unsigned char *, std::size_t,
                                              std::size_t, std::size_t, bool>;

   private:
    /**
      The event rewriter reference.
     */
    Database_rewrite &m_event_rewriter;

    /**
      Expands the buffer if needed.
    */
    std::tuple<unsigned char *, std::size_t, bool> reserve(
        unsigned char *buffer, std::size_t capacity, std::size_t size) {
      if (size > capacity) {
        auto outsize{size};
        outsize = round(((size + BINLOG_CHECKSUM_LEN) / 1024.0) + 1) * 1024;
        buffer = (unsigned char *)realloc(buffer, outsize);
        if (!buffer) {
          return std::make_tuple(nullptr, 0, true);
        }
        return std::make_tuple(buffer, outsize, false);
      } else
        return std::make_tuple(buffer, capacity, false);
    }

    class Buffer_realloc_manager {
     private:
      unsigned char **m_buffer{nullptr};

     public:
      explicit Buffer_realloc_manager(unsigned char **buffer)
          : m_buffer{buffer} {}
      ~Buffer_realloc_manager() {
        if (m_buffer != nullptr) free(*m_buffer);
      }
      void release() { m_buffer = nullptr; }
    };

    Rewrite_payload_result rewrite_inner_events(
        binary_log::transaction::compression::type compression_type,
        const char *orig_payload, std::size_t orig_payload_size,
        std::size_t orig_payload_uncompressed_size,
        const binary_log::Format_description_event &fde) {
      // to return error or not
      auto err{false};
      auto error_val = Rewrite_payload_result{nullptr, 0, 0, 0, true};

      // output variables
      unsigned char *obuffer{nullptr};
      std::size_t obuffer_size{0};
      std::size_t obuffer_capacity{0};
      std::size_t obuffer_size_uncompressed{0};

      // temporary buffer for holding uncompressed and rewritten events
      unsigned char *ibuffer{nullptr};
      std::size_t ibuffer_capacity{0};

      // RAII objects
      Buffer_realloc_manager obuffer_dealloc_guard(&obuffer);
      Buffer_realloc_manager ibuffer_dealloc_guard(&ibuffer);

      // iterator to decompress events
      binary_log::transaction::compression::Iterable_buffer it(
          orig_payload, orig_payload_size, orig_payload_uncompressed_size,
          compression_type);

      // compressor to compress this again
      auto compressor =
          binary_log::transaction::compression::Factory::build_compressor(
              compression_type);

      compressor->set_buffer(obuffer, obuffer_size);
      compressor->reserve(orig_payload_uncompressed_size);
      compressor->open();

      // rewrite and compress
      for (auto ptr : it) {
        std::size_t ev_len{uint4korr(ptr + EVENT_LEN_OFFSET)};

        // reserve input buffer size (we are modifying the input buffer contents
        // before compressing it back).
        std::tie(ibuffer, ibuffer_capacity, err) =
            reserve(ibuffer, ibuffer_capacity, ev_len);
        if (err) return error_val;
        memcpy(ibuffer, ptr, ev_len);

        // rewrite the database name if needed
        std::tie(ibuffer, ibuffer_capacity, ev_len, err) =
            m_event_rewriter.rewrite_event(ibuffer, ibuffer_capacity, ev_len,
                                           fde);
        if (err) return error_val;

        auto left{ev_len};
        while (left > 0 && !err) {
          auto pos{ibuffer + (ev_len - left)};
          std::tie(left, err) = compressor->compress(pos, left);
        }

        if (err) return error_val;
        obuffer_size_uncompressed += ev_len;
      }

      compressor->close();
      std::tie(obuffer, obuffer_size, obuffer_capacity) =
          compressor->get_buffer();

      // do not dispose of the obuffer (disable RAII for obuffer)
      obuffer_dealloc_guard.release();

      // set the new one and adjust event settings
      return Rewrite_payload_result{obuffer, obuffer_capacity, obuffer_size,
                                    obuffer_size_uncompressed, false};
    }

   public:
    explicit Transaction_payload_content_rewriter(Database_rewrite &rewriter)
        : m_event_rewriter(rewriter) {}

    /**
      This member function SHALL decompress, rewrite the contents of the
      payload event, compress it again and then re-encode it.

      @param buffer the buffer holding this event encoded.
      @param buffer_capacity the capacity of the buffer.
      @param fde The format description event to decode this event.

      @return a tuple with the result of the rewrite.
     */
    Rewrite_result rewrite_transaction_payload(
        unsigned char *buffer, std::size_t buffer_capacity,
        binary_log::Format_description_event const &fde) {
      assert(buffer[EVENT_TYPE_OFFSET] ==
             binary_log::TRANSACTION_PAYLOAD_EVENT);
      binary_log::Transaction_payload_event tpe((const char *)buffer, &fde);

      auto orig_payload{tpe.get_payload()};
      auto orig_payload_size{tpe.get_payload_size()};
      auto orig_payload_uncompressed_size{tpe.get_uncompressed_size()};
      auto orig_payload_compression_type{tpe.get_compression_type()};

      unsigned char *rewritten_payload{nullptr};
      std::size_t rewritten_payload_size{0};
      std::size_t rewritten_payload_capacity{0};
      std::size_t rewritten_payload_uncompressed_size{0};

      auto rewrite_payload_res{false};
      auto has_crc{fde.footer()->checksum_alg ==
                   binary_log::BINLOG_CHECKSUM_ALG_CRC32};

      // Rewrite its contents as needed
      std::tie(rewritten_payload, rewritten_payload_capacity,
               rewritten_payload_size, rewritten_payload_uncompressed_size,
               rewrite_payload_res) =
          rewrite_inner_events(orig_payload_compression_type, orig_payload,
                               orig_payload_size,
                               orig_payload_uncompressed_size, fde);

      if (rewrite_payload_res) return Rewrite_result{nullptr, 0, 0, true};

      // create a new TPE with the new buffer
      binary_log::Transaction_payload_event new_tpe(
          reinterpret_cast<const char *>(rewritten_payload),
          rewritten_payload_size, orig_payload_compression_type,
          rewritten_payload_uncompressed_size);

      // start encoding it
      auto codec =
          binary_log::codecs::Factory::build_codec(tpe.header()->type_code);
      uchar tpe_buffer[binary_log::Transaction_payload_event::MAX_DATA_LENGTH];
      auto result = codec->encode(new_tpe, tpe_buffer, sizeof(tpe_buffer));
      if (result.second == true) return Rewrite_result{nullptr, 0, 0, true};

      // Now adjust the event buffer itself
      auto new_data_size = result.first + rewritten_payload_size;
      auto new_event_size = LOG_EVENT_HEADER_LEN + new_data_size;
      if (has_crc) new_event_size += BINLOG_CHECKSUM_LEN;
      if (new_event_size > buffer_capacity)
        buffer = (unsigned char *)my_realloc(PSI_NOT_INSTRUMENTED, buffer,
                                             new_event_size, MYF(0));

      // now write everything into the event buffer
      auto ptr = buffer;

      // preserve the current event header, but adjust the event size
      int4store(ptr + EVENT_LEN_OFFSET, new_event_size);
      ptr += LOG_EVENT_HEADER_LEN;

      // add the new tpe header
      memmove(ptr, tpe_buffer, result.first);
      ptr += result.first;

      // add the new payload
      memmove(ptr, rewritten_payload, rewritten_payload_size);
      ptr += rewritten_payload_size;

      // now can free the new payload, as we have moved it to the
      // event buffer
      free(rewritten_payload);

      // recalculate checksum
      if (has_crc) {
        ha_checksum crc{0};
        uchar buf[BINLOG_CHECKSUM_LEN];
        crc = checksum_crc32(crc, buffer, new_event_size - BINLOG_CHECKSUM_LEN);
        int4store(buf, crc);
        memcpy(ptr, buf, sizeof(buf));
      }

      return Rewrite_result{buffer, new_event_size, new_event_size, false};
    }
  };

 protected:
  /**
    A map that establishes the relationship between from the source
    database name that is to be rewritten into the target one.

    The key of the map is the "from" database name. The value of the
    map is is the "to" database name that we are rewriting the
    name into.
   */
  std::map<std::string, std::string> m_dict;

  /**
    A special rewriter for those transactions that are enclosed in a
    Transaction_payload event.
   */
  std::unique_ptr<Transaction_payload_content_rewriter>
      m_transaction_payload_rewriter{nullptr};

  /**
    This function gets the offset in the buffer for the dbname and
    dbname length.

    @param buffer the event buffer
    @param buffer_size the event length
    @param fde the format description event to decode parts of this buffer

    @return a tuple containing:
            - dbname offset
            - dbname length offset
            - boolean specifying whether this is an event that needs rewrite
              checks
            - boolean specifying whether an error was found
  */
  std::tuple<my_off_t, my_off_t, bool, bool> get_dbname_and_dblen_offsets(
      const unsigned char *buffer, size_t buffer_size,
      binary_log::Format_description_event const &fde) {
    my_off_t off_dbname = 0;
    my_off_t off_dbname_len = 0;
    bool error = false;
    bool needs_rewrite_check = false;
    auto event_type = (Log_event_type)buffer[EVENT_TYPE_OFFSET];

    switch (event_type) {
      case binary_log::TABLE_MAP_EVENT: {
        /*
          Before rewriting:

          +-------------+-----------+----------+------+----------------+
          |common_header|post_header|old_db_len|old_db|event data...   |
          +-------------+-----------+----------+------+----------------+

          Note that table map log event uses only one byte for database length.

        */
        off_dbname_len = fde.common_header_len +
                         fde.post_header_len[binary_log::TABLE_MAP_EVENT - 1];
        off_dbname = off_dbname_len + 1;
        needs_rewrite_check = true;
      } break;
      case binary_log::EXECUTE_LOAD_QUERY_EVENT:
      case binary_log::QUERY_EVENT: {
        /*
          The QUERY_EVENT buffer structure:

          Before Rewriting :

            +-------------+-----------+-----------+------+------+
            |common_header|post_header|status_vars|old_db|...   |
            +-------------+-----------+-----------+------+------+

          After Rewriting :

            +-------------+-----------+-----------+------+------+
            |common_header|post_header|status_vars|new_db|...   |
            +-------------+-----------+-----------+------+------+

          The db_len is inside the post header, more specifically:

            +---------+---------+------+--------+--------+------+
            |thread_id|exec_time|db_len|err_code|status_vars_len|
            +---------+---------+------+--------+--------+------+

          Thence we need to change the post header and the payload,
          which is the one carrying the database name.

          In case the new database name is longer than the old database
          length, it will reallocate the buffer.
        */
        uint8 common_header_len = fde.common_header_len;
        uint8 query_header_len =
            fde.post_header_len[binary_log::QUERY_EVENT - 1];
        const unsigned char *ptr = buffer;
        uint sv_len = 0;

        DBUG_EXECUTE_IF("simulate_corrupt_event_len", buffer_size = 0;);
        /* Error if the event content is too small */
        if (buffer_size < (common_header_len + query_header_len)) {
          error = true;
          goto end;
        }

        /* Check if there are status variables in the event */
        if ((query_header_len -
             binary_log::Query_event::QUERY_HEADER_MINIMAL_LEN) > 0) {
          sv_len = uint2korr(ptr + common_header_len +
                             binary_log::Query_event::Q_STATUS_VARS_LEN_OFFSET);
        }

        /* now we have a pointer to the position where the database is. */
        off_dbname_len =
            common_header_len + binary_log::Query_event::Q_DB_LEN_OFFSET;
        off_dbname = common_header_len + query_header_len + sv_len;

        if (off_dbname_len > buffer_size || off_dbname > buffer_size) {
          error = true;
          goto end;
        }

        if (event_type == binary_log::EXECUTE_LOAD_QUERY_EVENT)
          off_dbname += Binary_log_event::EXECUTE_LOAD_QUERY_EXTRA_HEADER_LEN;
        needs_rewrite_check = true;
      } break;
      default:
        break;
    }

  end:
    return std::make_tuple(off_dbname, off_dbname_len, needs_rewrite_check,
                           error);
  }

  Rewrite_result rewrite_event(unsigned char *buffer, size_t buffer_capacity,
                               size_t data_size,
                               binary_log::Format_description_event const &fde,
                               bool recalculate_crc = false) {
    auto the_buffer{buffer};
    auto the_buffer_capacity{buffer_capacity};
    auto the_data_size{data_size};
    std::string from{};
    std::string to{};
    int64_t delta{0};
    unsigned char *dbname_ptr{nullptr};
    unsigned char *dbname_len_ptr{nullptr};

    bool error{false};
    bool needs_rewrite{false};
    size_t offset_dbname_len{0};
    size_t offset_dbname{0};
    uint8_t dbname_len{0};
    const char *dbname{nullptr};

    std::tie(offset_dbname, offset_dbname_len, needs_rewrite, error) =
        get_dbname_and_dblen_offsets(buffer, data_size, fde);
    if (error || !needs_rewrite) goto end;

    // build the "from"
    dbname_len = static_cast<uint8_t>(buffer[offset_dbname_len]);
    dbname = reinterpret_cast<const char *>(buffer + offset_dbname);
    from = std::string(dbname, dbname_len);

    // check if we need to continue
    if (!is_rewrite_needed(from)) goto end;

    // if we do, we need to find the name to rewrite to (the "to")
    to = m_dict[from];

    // need to adjust the buffer layout or even reallocate
    delta = to.size() - from.size();
    // need to reallocate
    if ((delta + data_size) > buffer_capacity) {
      the_buffer_capacity = buffer_capacity + delta;
      the_buffer = (unsigned char *)my_realloc(PSI_NOT_INSTRUMENTED, buffer,
                                               the_buffer_capacity, MYF(0));
      /* purecov: begin inspected */
      if (!the_buffer) {
        // OOM
        error = true;
        goto end;
      }
      /* purecov: end */
    }

    // adjust the size of the event
    the_data_size += delta;

    // need to move bytes around in the buffer if needed
    if (the_data_size != data_size) {
      unsigned char *to_tail_ptr = the_buffer + offset_dbname + to.size();
      unsigned char *from_tail_ptr = the_buffer + offset_dbname + from.size();
      size_t to_tail_size = data_size - (offset_dbname + from.size());

      // move the tail (so we do not risk overwriting it)
      memmove(to_tail_ptr, from_tail_ptr, to_tail_size);
    }

    dbname_ptr = the_buffer + offset_dbname;
    memcpy(dbname_ptr, to.c_str(), to.size());

    assert(to.size() < UINT8_MAX);
    dbname_len_ptr = the_buffer + offset_dbname_len;
    *dbname_len_ptr = (char)to.size();

    // Update event length in header.
    int4store(the_buffer + EVENT_LEN_OFFSET, the_data_size);

    // now recalculate the checksum
    if (recalculate_crc) {
      auto ptr = the_buffer + the_data_size - BINLOG_CHECKSUM_LEN;
      ha_checksum crc{};
      uchar buf[BINLOG_CHECKSUM_LEN];
      crc = checksum_crc32(crc, the_buffer, (ptr - the_buffer));
      int4store(buf, crc);
      memcpy(ptr, buf, sizeof(buf));
    }

  end:
    return std::make_tuple(the_buffer, the_buffer_capacity, the_data_size,
                           error);
  }

  /**
    This function shall return true if the event needs to be processed for
    rewriting the database.

    @param event_type the event type code.

    @return true if the database needs to be rewritten.
   */
  bool is_rewrite_needed_for_event(Log_event_type event_type) {
    switch (event_type) {
      case binary_log::TABLE_MAP_EVENT:
      case binary_log::EXECUTE_LOAD_QUERY_EVENT:
      case binary_log::QUERY_EVENT:
      case binary_log::TRANSACTION_PAYLOAD_EVENT:
        return true;
      default:
        return false;
    }
  }

 public:
  Database_rewrite() = default;

  ~Database_rewrite() { m_dict.clear(); }

  /**
    Shall register a rule to rewrite from one database name to another.
    @param from the database name to rewrite from.
    @param to the database name to rewrite to.
   */
  void register_rule(std::string from, std::string to) {
    m_dict.insert(std::pair<std::string, std::string>(from, to));
  }

  /**
    Shall unregister a rewrite rule for a given database. If the name is
    not registered, then no action is taken and no error reported.
    The name of database to be used in this invocation is the original
    database name.

    @param from the original database name used when the rewrite rule
                was registered.
   */
  void unregister_rule(std::string from) { m_dict.erase(from); }

  /**
    Returns true if this database name needs to be rewritten.

    @param dbname The database name.
    @return true if a database name rewrite is needed, false otherwise.
   */
  bool is_rewrite_needed(std::string dbname) {
    return !m_dict.empty() && m_dict.find(dbname) != m_dict.end();
  }

  /**
    Shall rewrite the database name in the given buffer. This function
    is called when rewriting events in raw_mode.

    @param buffer the full event still not decoded.
    @param buffer_capacity the event buffer size.
    @param data_size the size of the buffer filled with meaningful data.
    @param fde the format description event to decode the event.
    @param skip_transaction_payload_event Whether to skip the
                                          Transaction_payload_event or not

    @return a tuple containing:
            - A pointer to the buffer after the changes (if any).
            - The buffer capacity size updated.
            - The event data size.
            - A boolean specifying whether there was an error or not.
   */
  Rewrite_result rewrite_raw(unsigned char *buffer, size_t buffer_capacity,
                             size_t data_size,
                             binary_log::Format_description_event const &fde,
                             bool skip_transaction_payload_event = false) {
    assert(buffer_capacity >= data_size);
    auto event_type = (Log_event_type)buffer[EVENT_TYPE_OFFSET];
    if (m_dict.empty() || !is_rewrite_needed_for_event(event_type))
      return Rewrite_result{buffer, buffer_capacity, data_size, false};

    switch (event_type) {
      case binary_log::TRANSACTION_PAYLOAD_EVENT: {
        if (!skip_transaction_payload_event) {
          if (m_transaction_payload_rewriter == nullptr)
            m_transaction_payload_rewriter =
                std::make_unique<Transaction_payload_content_rewriter>(*this);
          return m_transaction_payload_rewriter->rewrite_transaction_payload(
              buffer, buffer_capacity, fde);
        } else
          return Rewrite_result{buffer, buffer_capacity, buffer_capacity,
                                false};
      }
      default: {
        bool recalculate_crc =
            fde.footer()->checksum_alg == binary_log::BINLOG_CHECKSUM_ALG_CRC32;
        return rewrite_event(buffer, buffer_capacity, data_size, fde,
                             recalculate_crc);
      }
    }
  }

  /**
    Rewrites the event database if needed. This function is called when
    rewriting events not in raw mode.

    @param buffer the full event still not decoded.
    @param buffer_capacity the event buffer size.
    @param data_size the size of the buffer filled with meaningful data.
    @param fde the format description event to decode the event.

    @return a tuple with the pointer to the buffer with the database rewritten,
            the rewritten buffer capacity, the rewritten buffer meaningful
            bytes, and whether there was an error or not.
   */
  Rewrite_result rewrite(unsigned char *buffer, size_t buffer_capacity,
                         size_t data_size,
                         binary_log::Format_description_event const &fde) {
    return rewrite_raw(buffer, buffer_capacity, data_size, fde, true);
  }
};

/**
  The database rewriter handler for Table map and Query log events.
 */
Database_rewrite global_database_rewriter;

/*
  The character set used should be equal to the one used in mysqld.cc for
  server rewrite-db
*/
#define mysqld_charset &my_charset_latin1

#define CLIENT_CAPABILITIES \
  (CLIENT_LONG_PASSWORD | CLIENT_LONG_FLAG | CLIENT_LOCAL_FILES)

char server_version[SERVER_VERSION_LENGTH];
ulong filter_server_id = 0;

/*
  This structure is used to store the event and the log position of the events
  which is later used to print the event details from correct log positions.
  The Log_event *event is used to store the pointer to the current event and
  the event_pos is used to store the current event log position.
*/
struct buff_event_info {
  Log_event *event = nullptr;
  my_off_t event_pos = 0;
};

/*
  One statement can result in a sequence of several events: Intvar_log_events,
  User_var_log_events, and Rand_log_events, followed by one
  Query_log_event. If statements are filtered out, the filter has to be
  checked for the Query_log_event. So we have to buffer the Intvar,
  User_var, and Rand events and their corresponding log positions until we see
  the Query_log_event. This dynamic array buff_ev is used to buffer a structure
  which stores such an event and the corresponding log position.
*/
typedef Prealloced_array<buff_event_info, 16> Buff_ev;
Buff_ev *buff_ev{nullptr};

// needed by net_serv.c
ulong bytes_sent = 0L, bytes_received = 0L;
ulong mysqld_net_retry_count = 10L;
ulong open_files_limit;
ulong opt_binlog_rows_event_max_size;
uint test_flags = 0;
static uint opt_protocol = 0;
static uint opt_compress = 0;
static uint opt_print_sql_string = 0;
static uint opt_use_dscp = 0;
static FILE *result_file;

#ifndef NDEBUG
static const char *default_dbug_option = "d:t:o,/tmp/mysqlbinlog.trace";
#endif
static const char *load_default_groups[] = {"mysqlbinlog", "client", nullptr};

static bool one_database = false, disable_log_bin = false;
static bool opt_hexdump = false;
const char *base64_output_mode_names[] = {"NEVER", "AUTO", "UNSPEC",
                                          "DECODE-ROWS", NullS};
TYPELIB base64_output_mode_typelib = {
    array_elements(base64_output_mode_names) - 1, "", base64_output_mode_names,
    nullptr};
static enum_base64_output_mode opt_base64_output_mode = BASE64_OUTPUT_UNSPEC;
static char *opt_base64_output_mode_str = nullptr;
static bool opt_remote_alias = false;
const char *remote_proto_names[] = {"BINLOG-DUMP-NON-GTIDS",
                                    "BINLOG-DUMP-GTIDS", NullS};
TYPELIB remote_proto_typelib = {array_elements(remote_proto_names) - 1, "",
                                remote_proto_names, nullptr};
static enum enum_remote_proto {
  BINLOG_DUMP_NON_GTID = 0,
  BINLOG_DUMP_GTID = 1,
  BINLOG_LOCAL = 2
} opt_remote_proto = BINLOG_LOCAL;
static char *opt_remote_proto_str = nullptr;
static char *database = nullptr;
static char *output_file = nullptr;
static char *rewrite = nullptr;
bool force_opt = false, short_form = false, idempotent_mode = false;
int mta_workers = 0;
static bool debug_info_flag, debug_check_flag;
static bool force_if_open_opt = true, raw_mode = false;
static bool to_last_remote_log = false, stop_never = false;
static bool opt_verify_binlog_checksum = true;
static bool opt_skip_verify_if_open = true;
static ulonglong offset = 0;
static int64 stop_never_slave_server_id = -1;
static int64 connection_server_id = -1;
static char *host = nullptr;
static int port = 0;
static uint my_end_arg;
static const char *sock = nullptr;
static char *opt_plugin_dir = nullptr, *opt_default_auth = nullptr;

#if defined(_WIN32)
static char *shared_memory_base_name = nullptr;
#endif
static char *user = nullptr;
static char *pass = nullptr;
static char *opt_bind_addr = nullptr;
static char *charset = nullptr;

static uint verbose = 0;

static ulonglong start_position, stop_position;
#define start_position_mot ((my_off_t)start_position)
#define stop_position_mot ((my_off_t)stop_position)

static char *start_datetime_str, *stop_datetime_str;
static my_time_t start_datetime = 0, stop_datetime = MYTIME_MAX_VALUE;
static ulonglong rec_count = 0;
static MYSQL *mysql = nullptr;
static char *dirname_for_local_load = nullptr;
static uint opt_server_id_bits = 0;
ulong opt_server_id_mask = 0;
Sid_map *global_sid_map = nullptr;
Checkable_rwlock *global_sid_lock = nullptr;
Gtid_set *gtid_set_included = nullptr;
Gtid_set *gtid_set_excluded = nullptr;
Dbtid_set dbtid_set_excluded;
static uint opt_zstd_compress_level = default_zstd_compression_level;
static char *opt_compress_algorithm = nullptr;
Gtid_set *gtid_set_stop = nullptr;
static bool opt_print_gtids = false;

static bool opt_print_table_metadata;

/**
 * Used for --opt-skip-empty-trans
 */
static Log_event *begin_query_ev_cache = nullptr;
static string cur_database = "";

/**
 * Used for --opt-read-from-binlog-server
 */
static const std::string binlog_server_finish_err_msg =
    "The binlog server has finished sending all available binlogs from the "
    "HDFS and has no more binlogs to send.";

enum class Check_database_decision {
  EMPTY_EVENT_DATABASE = 2,
  CHANGED = 1,
  OK = 0,
  ERROR = -1
};

/**
  Exit status for functions in this file.
*/
enum Exit_status {
  /** No error occurred and execution should continue. */
  OK_CONTINUE = 0,
  /** An error occurred and execution should stop. */
  ERROR_STOP,
  /** No error occurred but execution should stop. */
  OK_STOP
};

/*
  Options that will be used to filter out events.
*/
static char *opt_include_gtids_str = nullptr, *opt_exclude_gtids_str = nullptr,
            *opt_include_gtids_from_file_str = nullptr,
            *opt_exclude_gtids_from_file_str = nullptr,
            *opt_start_gtid_str = nullptr, *opt_find_gtid_str = nullptr,
            *opt_stop_gtid_str = nullptr;
static char *opt_exclude_dbtids_str = nullptr;
static char *opt_index_file_str = nullptr;
Gtid_set_map previous_gtid_set_map;
static bool opt_skip_gtids = false;
static bool opt_skip_rows_query =
    false; /* Placeholder for skip_rows_query diff */
static bool opt_skip_empty_trans = 0;
static bool opt_read_from_binlog_server = 0;
static bool filter_based_on_gtids = false;
static bool filter_based_on_dbtids = false;
static bool opt_require_row_format = false;

/* It is set to true when BEGIN is found, and false when the transaction ends.
 */
static bool in_transaction = false;
/* It is set to true when GTID is found, and false when the transaction ends. */
static bool seen_gtid = false;

static uint opt_receive_buffer_size = 0;

static Exit_status dump_local_log_entries(PRINT_EVENT_INFO *print_event_info,
                                          const char *logname);
static Exit_status dump_remote_log_entries(PRINT_EVENT_INFO *print_event_info,
                                           const char *logname);
static Exit_status dump_single_log(PRINT_EVENT_INFO *print_event_info,
                                   const char *logname);
static Exit_status dump_multiple_logs(int argc, char **argv);
static Exit_status safe_connect();

struct buff_event_info buff_event;

// the last seen rows_query event buffered so we can check database/table
// filters in the table_map/query event before printing it
static buff_event_info last_rows_query_event;
// the `temp_buf` data member of the last seen rows_query event, this stores the
// serialized event and is used to print the event in base64 format
static char *last_rows_query_event_temp_buf = nullptr;

class Load_log_processor {
  char target_dir_name[FN_REFLEN];
  size_t target_dir_name_len;

  /*
    When we see first event corresponding to some LOAD DATA statement in
    binlog, we create temporary file to store data to be loaded.
    We add name of this file to file_names set using its file_id as index.
  */
  struct File_name_record {
    char *fname;
  };

  typedef std::map<uint, File_name_record> File_names;
  File_names file_names;

  /**
    Looks for a non-existing filename by adding a numerical suffix to
    the given base name, creates the generated file, and returns the
    filename by modifying the filename argument.

    @param[in,out] filename Base filename

    @param[in,out] file_name_end Pointer to last character of
    filename.  The numerical suffix will be written to this position.
    Note that there must be a least five bytes of allocated memory
    after file_name_end.

    @retval -1 Error (can't find new filename).
    @retval >=0 Found file.
  */
  File create_unique_file(char *filename, char *file_name_end) {
    File res;
    /* If we have to try more than 1000 times, something is seriously wrong */
    for (uint version = 0; version < 1000; version++) {
      sprintf(file_name_end, "-%x", version);
      if ((res = my_create(filename, 0, O_CREAT | O_EXCL | O_WRONLY, MYF(0))) !=
          -1)
        return res;
    }
    return -1;
  }

 public:
  Load_log_processor() : file_names() {}
  ~Load_log_processor() = default;

  void init_by_dir_name(const char *dir) {
    target_dir_name_len =
        (convert_dirname(target_dir_name, dir, NullS) - target_dir_name);
  }
  void init_by_cur_dir() {
    if (my_getwd(target_dir_name, sizeof(target_dir_name), MYF(MY_WME)))
      exit(1);
    target_dir_name_len = strlen(target_dir_name);
  }
  void destroy() {
    File_names::iterator iter = file_names.begin();
    File_names::iterator end = file_names.end();
    for (; iter != end; ++iter) {
      File_name_record *ptr = &iter->second;
      if (ptr->fname) {
        my_free(ptr->fname);
        memset(ptr, 0, sizeof(File_name_record));
      }
    }

    file_names.clear();
  }

  /**
    Obtain file name of temporary file for LOAD DATA statement by its
    file_id and remove it from this Load_log_processor's list of events.

    @param[in] file_id Identifier for the LOAD DATA statement.

    Checks whether we have already seen Begin_load_query event for
    this file_id. If yes, returns the file name of the corresponding
    temporary file and removes the filename from the array of active
    temporary files.  From this moment, the caller is responsible for
    freeing the memory occupied by this name.

    @return String with the name of the temporary file, or NULL if we
    have not seen any Begin_load_query_event with this file_id.
  */
  char *grab_fname(uint file_id) {
    File_name_record *ptr;
    char *res = nullptr;

    File_names::iterator it = file_names.find(file_id);
    if (it == file_names.end()) return nullptr;
    ptr = &((*it).second);
    res = ptr->fname;
    memset(ptr, 0, sizeof(File_name_record));
    return res;
  }
  Exit_status process(Begin_load_query_log_event *ce);
  Exit_status process(Append_block_log_event *ae);
  Exit_status process_first_event(const char *bname, size_t blen,
                                  const uchar *block, size_t block_len,
                                  uint file_id);
};

/**
  Process the first event in the sequence of events representing a
  LOAD DATA statement.

  Creates a temporary file to be used in LOAD DATA and writes first block of
  data to it. Registers its file name in the array of active temporary files.

  @param bname Base name for temporary file to be created.
  @param blen Base name length.
  @param block First block of data to be loaded.
  @param block_len First block length.
  @param file_id Identifies the LOAD DATA statement.
  this type of event.

  @retval ERROR_STOP An error occurred - the program should terminate.
  @retval OK_CONTINUE No error, the program should continue.
*/
Exit_status Load_log_processor::process_first_event(const char *bname,
                                                    size_t blen,
                                                    const uchar *block,
                                                    size_t block_len,
                                                    uint file_id) {
  size_t full_len = target_dir_name_len + blen + 9 + 9 + 1;
  Exit_status retval = OK_CONTINUE;
  char *fname, *ptr;
  File file;
  File_name_record rec;
  DBUG_TRACE;

  if (!(fname =
            (char *)my_malloc(PSI_NOT_INSTRUMENTED, full_len, MYF(MY_WME)))) {
    error("Out of memory.");
    return ERROR_STOP;
  }