slave_log_event.cpp

/* Copyright 2011 ZAO "Begun".
 *
 * This library is free software; you can redistribute it and/or modify it under
 * the terms of the GNU Lesser General Public License as published by the Free
 * Software Foundation; either version 3 of the License, or (at your option)
 * any later version.
 * This library 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 Lesser General Public License for more
 * details.
 * You should have received a copy of the GNU Lesser General Public License
 * along with this library.  If not, see <http://www.gnu.org/licenses/>.
*/

#include <string>
#include <vector>
#include <map>
#include <set>

#include <my_byteorder.h>
#undef min
#undef max
#undef test

#include <mysql.h>

#include <zlib.h>

#include "relayloginfo.h"
#include "slave_log_event.h"

#include "SlaveStats.h"
#include "Logging.h"


namespace
{
void bin2hex_nz(char* dst, const uint8_t* src, size_t sz_src)
{
    if (!src || !dst) return;

    static const char* hex = "0123456789abcdef";

    for (size_t i = 0; i < sz_src; ++i)
    {
        *dst++ = hex[src[i] >> 4];
        *dst++ = hex[src[i] & 0x0f];
    }
}

std::string bin2hex(const uint8_t* src, size_t sz_src)
{
    std::string res;
    res.resize(sz_src * 2);
    bin2hex_nz(const_cast<char*>(res.data()), src, sz_src);
    return res;
}
} // namespace anonymous

namespace slave {



//---------------------------------------------------------------------------------------

void Basic_event_info::parse(const char* _buf, unsigned int _event_len) {

    buf = _buf;
    event_len = _event_len;

    if (event_len < LOG_POS_OFFSET + 4) {
        LOG_ERROR(log, "Sanity check failed: " << event_len << " " << LOG_POS_OFFSET + 4);
        throw std::runtime_error("Basic_event_info::parse failed");
    }

    type = (slave::Log_event_type)buf[EVENT_TYPE_OFFSET];

    when = uint4korr(buf);
    server_id = uint4korr(buf + SERVER_ID_OFFSET);
    log_pos = uint4korr(buf + LOG_POS_OFFSET);
}

Rotate_event_info::Rotate_event_info(const char* buf, unsigned int event_len) {

    if (event_len < LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN) {
        LOG_ERROR(log, "Sanity check failed: " << event_len << " " << LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN);
        throw std::runtime_error("Rotate_event_info::Rotate_event_info failed");
    }

    pos = uint8korr(buf + LOG_EVENT_HEADER_LEN + R_POS_OFFSET);
    ident_len = (event_len - LOG_EVENT_HEADER_LEN - ROTATE_HEADER_LEN);
    new_log_ident.assign(buf + LOG_EVENT_HEADER_LEN + ROTATE_HEADER_LEN, ident_len);
}


Query_event_info::Query_event_info(const char* buf, unsigned int event_len) {

    if (event_len < LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN) {
        LOG_ERROR(log, "Sanity check failed: " << event_len << " " << LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN);
        throw std::runtime_error("Query_event_info::Query_event_info failed");
    }

    unsigned int db_len = (unsigned int)buf[LOG_EVENT_HEADER_LEN + Q_DB_LEN_OFFSET];

    unsigned int status_vars_len = uint2korr(buf + LOG_EVENT_HEADER_LEN + Q_STATUS_VARS_LEN_OFFSET);

    size_t data_len = event_len - (LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN) - status_vars_len;

    db_name.assign(buf + LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN + status_vars_len, db_len);
    query.assign(buf + LOG_EVENT_HEADER_LEN + QUERY_HEADER_LEN + status_vars_len + db_len + 1,
                 data_len - db_len - 1);
}


Table_map_event_info::Table_map_event_info(const char* buf, unsigned int event_len) {

    if (event_len < LOG_EVENT_HEADER_LEN + TABLE_MAP_HEADER_LEN + 2) {
        LOG_ERROR(log, "Sanity check failed: " << event_len << " " << LOG_EVENT_HEADER_LEN + TABLE_MAP_HEADER_LEN + 2);
        throw std::runtime_error("Table_map_event_info::Table_map_event_info failed");
    }

    m_table_id = uint6korr(buf + LOG_EVENT_HEADER_LEN + TM_MAPID_OFFSET);

    unsigned char* p_dblen = (unsigned char*)(buf + LOG_EVENT_HEADER_LEN + TABLE_MAP_HEADER_LEN);
    size_t dblen = *(p_dblen);

    m_dbnam.assign((const char*)(p_dblen + 1), dblen);

    unsigned char* p_tblen = p_dblen + dblen + 2;
    size_t tblen = *(p_tblen);

    m_tblnam.assign((const char*)(p_tblen + 1), tblen);

    unsigned char* p_width = p_tblen + tblen + 2;
    unsigned long width = net_field_length(&p_width);

    m_cols_types.assign(p_width, p_width + width);

    unsigned char *metadata = p_width + width;
    unsigned long metadata_length = net_field_length(&metadata);

    m_metadata.assign(metadata, metadata + metadata_length);
}

Row_event_info::Row_event_info(const char* buf, const unsigned int event_len, const bool is_update, const bool is_v2_event)
{
    unsigned header_len = is_v2_event ? ROWS_HEADER_LEN : ROWS_HEADER_LEN_V1;

    if (event_len < LOG_EVENT_HEADER_LEN + header_len + 2) {
        LOG_ERROR(log, "Sanity check failed: " << event_len << " " << LOG_EVENT_HEADER_LEN + header_len + 2);
        throw std::runtime_error("Row_event_info::Row_event_info() failed - event_len too small");
    }

    has_after_image = is_update;
    unsigned char* p_header = (unsigned char*)buf + LOG_EVENT_HEADER_LEN;

    p_header += ROWS_MAPID_OFFSET;
    m_table_id = uint6korr(p_header);
    p_header += ROWS_FLAGS_OFFSET + 2; // skip flags

    if (is_v2_event) {
        const unsigned var_header_len = uint2korr(p_header);

        if (var_header_len < 2) {
            throw std::runtime_error("Row_event_info::Row_event_info() failed - var_header_len < 2");
        }

        header_len += var_header_len - 2;
    }

    unsigned char* p_data = (unsigned char*)buf + LOG_EVENT_HEADER_LEN + header_len;

    m_width = net_field_length(&p_data);
    m_cols.assign(p_data, p_data + (m_width + 7) / 8);
    p_data += m_cols.size();

    if (is_update) {
        m_cols_ai.assign(p_data, p_data + m_cols.size());
        p_data += m_cols_ai.size();
    }

    m_rows_buf = p_data;
    m_rows_end = (unsigned char*)buf + event_len;
}

Gtid_event_info::Gtid_event_info(const char* buf, unsigned int event_len)
{
    if (event_len < LOG_EVENT_HEADER_LEN + GTID_EVENT_LEN) {
        LOG_ERROR(log, "Sanity check failed: " << event_len << " " << LOG_EVENT_HEADER_LEN + GTID_EVENT_LEN);
        throw std::runtime_error("Gtid_event_info::Gtid_event_info failed");
    }

    m_sid = bin2hex((uchar*)buf + LOG_EVENT_HEADER_LEN + ENCODED_FLAG_LENGTH, ENCODED_SID_LENGTH);
    m_gno = sint8korr(buf + LOG_EVENT_HEADER_LEN + ENCODED_FLAG_LENGTH + ENCODED_SID_LENGTH);
}

/////////////////////////


inline void check_format_description_postlen(unsigned char* b, slave::Log_event_type type, unsigned char len) {

    if (b[(int)type - 1] != len) {

        LOG_ERROR(log, "Invalid Format_description event: event type " << (int)type
                  << " len: " << (int)b[(int)type - 1] << " != " << (int)len);

        throw std::runtime_error("Invalid Format_description event");
    }
}


inline void check_format_description(const char* buf, unsigned int event_len, bool master_ge_56) {

    buf += LOG_EVENT_MINIMAL_HEADER_LEN;

    uint16_t binlog_version;
    memcpy(&binlog_version, buf + ST_BINLOG_VER_OFFSET, ST_BINLOG_VER_LEN);
    if (4 != binlog_version)
    {
        LOG_ERROR(log, "Invalid binlog version: " << binlog_version << " != 4");
        throw std::runtime_error("Invalid binlog version");
    }

    size_t common_header_len = (unsigned char)(buf[ST_COMMON_HEADER_LEN_OFFSET]);

    if (common_header_len != LOG_EVENT_HEADER_LEN) {

        LOG_ERROR(log, "Invalid Format_description event: common_header_len " << common_header_len
                  << " != " << LOG_EVENT_HEADER_LEN);
        throw std::runtime_error("Invalid Format_description event");
    }

    // Check that binlog contains different event types not more than we know
    // If less - it's ok, backward compatibility
    const size_t number_of_event_types =
        event_len - (LOG_EVENT_MINIMAL_HEADER_LEN + ST_COMMON_HEADER_LEN_OFFSET + 1);

    if (number_of_event_types > LOG_EVENT_TYPES)
    {
        LOG_ERROR(log, "Invalid Format_description event: number_of_event_types " << number_of_event_types
                  << " > " << LOG_EVENT_TYPES);
        throw std::runtime_error("Invalid Format_description event");
    }

    unsigned char event_lens[LOG_EVENT_TYPES] = { 0, };

    ::memcpy(&event_lens[0], (unsigned char*)(buf + ST_COMMON_HEADER_LEN_OFFSET + 1), LOG_EVENT_TYPES);

    check_format_description_postlen(event_lens, XID_EVENT, 0);
    check_format_description_postlen(event_lens, QUERY_EVENT, QUERY_HEADER_LEN);
    check_format_description_postlen(event_lens, ROTATE_EVENT, ROTATE_HEADER_LEN);
    check_format_description_postlen(event_lens, FORMAT_DESCRIPTION_EVENT, START_V3_HEADER_LEN + 1 + number_of_event_types);
    check_format_description_postlen(event_lens, TABLE_MAP_EVENT, TABLE_MAP_HEADER_LEN);
    check_format_description_postlen(event_lens, WRITE_ROWS_EVENT_V1, ROWS_HEADER_LEN_V1);
    check_format_description_postlen(event_lens, UPDATE_ROWS_EVENT_V1, ROWS_HEADER_LEN_V1);
    check_format_description_postlen(event_lens, DELETE_ROWS_EVENT_V1, ROWS_HEADER_LEN_V1);
    if (master_ge_56)
    {
        check_format_description_postlen(event_lens, WRITE_ROWS_EVENT, ROWS_HEADER_LEN);
        check_format_description_postlen(event_lens, UPDATE_ROWS_EVENT, ROWS_HEADER_LEN);
        check_format_description_postlen(event_lens, DELETE_ROWS_EVENT, ROWS_HEADER_LEN);
    }

}


inline uint32_t checksum_crc32(uint32_t crc, const unsigned char* pos, size_t length)
{
    return static_cast<uint32_t>(::crc32(static_cast<unsigned int>(crc), pos, static_cast<unsigned int>(length)));
}


bool read_log_event(const char* buf, uint event_len, Basic_event_info& bei, EventStatIface* event_stat, bool master_ge_56, MasterInfo& master_info)

{

    bei.parse(buf, event_len);


    /* Check the integrity */

    if (event_len < EVENT_LEN_OFFSET ||
        bei.type >= ENUM_END_EVENT ||
        (uint) event_len != uint4korr(buf+EVENT_LEN_OFFSET))
    {
        LOG_ERROR(log, "Sanity check failed: " << event_len);
        throw std::runtime_error("slave::read_log_event failed");
    }

    if (master_ge_56 && bei.type == FORMAT_DESCRIPTION_EVENT)
    {
        enum_binlog_checksum_alg alg = static_cast<enum_binlog_checksum_alg>(*(buf + event_len - BINLOG_CHECKSUM_LEN - BINLOG_CHECKSUM_ALG_DESC_LEN));
        if (alg == BINLOG_CHECKSUM_ALG_OFF || alg == BINLOG_CHECKSUM_ALG_CRC32)
            master_info.checksum_alg = alg;
    }

    if (master_info.checksumEnabled())
    {
        uint32_t incoming;
        ::memcpy(&incoming, buf + event_len - BINLOG_CHECKSUM_LEN, sizeof(incoming));
        incoming = le32toh(incoming);

        uint32_t computed = checksum_crc32(0L, nullptr, 0);
        computed = checksum_crc32(computed, (const unsigned char*)buf, event_len - BINLOG_CHECKSUM_LEN);

        if (incoming != computed)
        {
            LOG_ERROR(log, "CRC32 check failed: incoming (" << incoming << ") != computed (" << computed << ")");
            throw std::runtime_error("slave::read_log_event failed");
        }
        bei.event_len -= BINLOG_CHECKSUM_LEN;
    }

    if (event_stat)
        if (bei.type != FORMAT_DESCRIPTION_EVENT && bei.type != ROTATE_EVENT &&
            bei.type != HEARTBEAT_LOG_EVENT && bei.type != PREVIOUS_GTIDS_LOG_EVENT)
            event_stat->tick(bei.when);

    switch (bei.type) {

    case FORMAT_DESCRIPTION_EVENT:

        if (master_ge_56)
            event_len -= (BINLOG_CHECKSUM_ALG_DESC_LEN + BINLOG_CHECKSUM_LEN);
        check_format_description(buf, event_len, master_ge_56);

        if (event_stat)
            event_stat->tickFormatDescription();
        return true;
        break;

    case QUERY_EVENT:
        if (event_stat)
            event_stat->tickQuery();
        return true;
    case ROTATE_EVENT:
        if (event_stat)
            event_stat->tickRotate();
        return true;
    case XID_EVENT:
        if (event_stat)
            event_stat->tickXid();
        return true;
    case WRITE_ROWS_EVENT_V1:
    case UPDATE_ROWS_EVENT_V1:
    case DELETE_ROWS_EVENT_V1:
    case WRITE_ROWS_EVENT:
    case UPDATE_ROWS_EVENT:
    case DELETE_ROWS_EVENT:
        // event_stat->tickModify is called from other place.
        return true;
    case TABLE_MAP_EVENT:
        // event_stat->processTableMapEvent is called from other place.
        return true;
        break;
    case GTID_LOG_EVENT:
        return true;
    case LOAD_EVENT:
    case NEW_LOAD_EVENT:
    case SLAVE_EVENT: /* can never happen (unused event) */
    case CREATE_FILE_EVENT:
    case APPEND_BLOCK_EVENT:
    case DELETE_FILE_EVENT:
    case EXEC_LOAD_EVENT:
    case START_EVENT_V3: /* this is sent only by MySQL <=4.x */
    case STOP_EVENT:
    case INTVAR_EVENT:
    case RAND_EVENT:
    case USER_VAR_EVENT:
    case PRE_GA_WRITE_ROWS_EVENT:
    case PRE_GA_UPDATE_ROWS_EVENT:
    case PRE_GA_DELETE_ROWS_EVENT:
    case BEGIN_LOAD_QUERY_EVENT:
    case EXECUTE_LOAD_QUERY_EVENT:
    case INCIDENT_EVENT:
    case HEARTBEAT_LOG_EVENT:
    case IGNORABLE_LOG_EVENT:
    case ROWS_QUERY_LOG_EVENT:
    case ANONYMOUS_GTID_LOG_EVENT:
    case PREVIOUS_GTIDS_LOG_EVENT:
    case TRANSACTION_CONTEXT_EVENT:
    case VIEW_CHANGE_EVENT:
    case XA_PREPARE_LOG_EVENT:
        if (event_stat)
            event_stat->tickOther();
        return false;
        break;

    default:
        LOG_ERROR( log, "Unknown event code: " << (int) bei.type);
        if (event_stat)
            event_stat->tickOther();
        return false;
        break;
    }

    //
    //

    if ( event_len > 1024*1024*3 ) { // >3 Mb
        LOG_ERROR(log, "Warning: event size > 3MB! Maybe a corrupted event!");
    }

    return true;
}



/*
 *
 */


size_t n_set_bits(const std::vector<unsigned char>& b, unsigned int count) {

    size_t ret = 0;

    for (unsigned int i = 0; i < count; ++i) {

        if (b[i / 8] & (1 << (i & 7)))
            ret++;
    }

    return ret;
}

template <typename T>
void fill_row(const slave::Table& table, T& row, unsigned index, const slave::FieldValue& value);

template <>
void fill_row<slave::Row>(const slave::Table& table, slave::Row& row, unsigned index, const slave::FieldValue& value)
{
    const auto& field = table.fields[index];
    if (table.column_filter.empty() || table.column_filter[index / 8] & (1 << (index & 7)))
        row[field->getFieldName()] = std::make_pair(field->field_type, value);
}

template <>
void fill_row<slave::RowVector>(const slave::Table& table, slave::RowVector& row, unsigned index, const slave::FieldValue& value)
{
    const auto& field = table.fields[index];
    if (table.column_filter.empty())
        row.emplace_back(field->field_type, value);
    else if (table.column_filter[index / 8] & (1 << (index & 7)))
        row[table.column_filter_fields[index]] = std::make_pair(field->field_type, value);
}

template <typename T>
void reserve_row(const slave::Table& table, T& row) {}

template <>
void reserve_row<slave::RowVector>(const slave::Table& table, slave::RowVector& row)
{
    if (table.column_filter.empty())
        row.reserve(table.fields.size());
    else
        row.resize(table.column_filter_count);
}

template <typename T>
unsigned char* unpack_row(const slave::Table& table,
                          T& _row,
                          unsigned int colcnt,
                          unsigned char* row,
                          const std::vector<unsigned char>& cols)
{

    LOG_TRACE(log, "Unpacking row: " << "fields in the table " << table.fields.size() << ", fields in the event " << colcnt
             << ", bitfield width " << cols.size());

    if (colcnt != table.fields.size()) {
        LOG_ERROR(log, "Field count mismatch in unpacking row for "
                  << table.full_name << ": " << colcnt << " != " << table.fields.size());
        throw std::runtime_error("unpack_row failed");
    }


    // pointer to start of data; skip master_null_bytes

    size_t master_null_byte_count = (n_set_bits(cols, colcnt) + 7) / 8;

    unsigned char* ptr = row + master_null_byte_count;

    //
    unsigned char* null_ptr = row;
    unsigned int null_mask = 1U;
    unsigned char null_bits = *null_ptr++;

    reserve_row<T>(table, _row);

    for (unsigned i = 0; i < colcnt; i++)
    {
        const auto& field = table.fields[i];

        if (!cols.empty() && !(cols[i / 8] & (1 << (i & 7)))) {

            LOG_TRACE(log, "field " << field->getFieldName() << " is not in column list.");
            continue;
        }

        if ((null_mask & 0xFF) == 0) {
            null_mask = 1U;
            null_bits = *null_ptr++;

            LOG_TRACE(log, "null mask reset.");
        }

        if (null_bits & null_mask) {

            LOG_TRACE(log, "field with NULL value found");

            // We don't unpack the field if it was NULL,
            // and put empty slave::FieldValue value to slave::Row's value
            // in order to indicate presence of NULL value.

            fill_row<T>(table, _row, i, nullFieldValue());
        }
        else
        {
            // We unpack the field to some certain value if it was NOT NULL
            ptr = (unsigned char*)field->unpack((const char*)ptr);
            fill_row<T>(table, _row, i, field->field_data);
        }

        null_mask <<= 1;

        LOG_TRACE(log, "field: " << field->getFieldName());

    }

    return ptr;
}


unsigned char* do_writedelete_row(const slave::Table& table,
                                  const Basic_event_info& bei,
                                  const Row_event_info& roi,
                                  unsigned char* row_start,
                                  ExtStateIface &ext_state) {

    slave::RecordSet _record_set;

    unsigned char* t = nullptr;
    if (table.row_type == RowType::Map)
        t = unpack_row(table, _record_set.m_row, roi.m_width, row_start, roi.m_cols);
    else
        t = unpack_row(table, _record_set.m_row_vec, roi.m_width, row_start, roi.m_cols);

    if (t == NULL) {
        return NULL;
    }

    _record_set.row_type = table.row_type;
    _record_set.when = bei.when;
    _record_set.tbl_name = table.table_name;
    _record_set.db_name = table.database_name;
    _record_set.type_event = (bei.type == WRITE_ROWS_EVENT_V1 || bei.type == WRITE_ROWS_EVENT ? slave::RecordSet::Write : slave::RecordSet::Delete);
    _record_set.master_id = bei.server_id;

    table.call_callback(_record_set, ext_state);

    return t;
}

unsigned char* do_update_row(const slave::Table& table,
                             const Basic_event_info& bei,
                             const Row_event_info& roi,
                             unsigned char* row_start,
                             ExtStateIface &ext_state) {

    slave::RecordSet _record_set;

    unsigned char* t = nullptr;
    if (table.row_type == RowType::Map)
        t = unpack_row(table, _record_set.m_old_row, roi.m_width, row_start, roi.m_cols);
    else
        t = unpack_row(table, _record_set.m_old_row_vec, roi.m_width, row_start, roi.m_cols);

    if (t == NULL) {
        return NULL;
    }

    if (table.row_type == RowType::Map)
        t = unpack_row(table, _record_set.m_row, roi.m_width, t, roi.m_cols_ai);
    else
        t = unpack_row(table, _record_set.m_row_vec, roi.m_width, t, roi.m_cols_ai);

    if (t == NULL) {
        return NULL;
    }

    _record_set.row_type = table.row_type;
    _record_set.when = bei.when;
    _record_set.tbl_name = table.table_name;
    _record_set.db_name = table.database_name;
    _record_set.type_event = slave::RecordSet::Update;
    _record_set.master_id = bei.server_id;

    table.call_callback(_record_set, ext_state);

    return t;
}

namespace // anonymous
{
    inline EventKind eventKind(Log_event_type type)
    {
        switch(type)
        {
        case WRITE_ROWS_EVENT_V1:
        case WRITE_ROWS_EVENT:    return eInsert;
        case UPDATE_ROWS_EVENT_V1:
        case UPDATE_ROWS_EVENT:   return eUpdate;
        case DELETE_ROWS_EVENT_V1:
        case DELETE_ROWS_EVENT:   return eDelete;
        default: throw std::logic_error("is not processable kind");
        }
    }

    typedef uint64_t time_stamp;

    inline time_stamp now()
    {
        timespec ts;
        clock_gettime(CLOCK_REALTIME, &ts);
        return ts.tv_sec * 1000000000 + ts.tv_nsec;
    }

} // namespace anonymous


void apply_row_event(const slave::RelayLogInfo& rli, const Basic_event_info& bei, const Row_event_info& roi, ExtStateIface& ext_state, EventStatIface* event_stat) {
    EventKind kind = eventKind(bei.type);
    std::pair<std::string,std::string> key = rli.getTableNameById(roi.m_table_id);

    LOG_DEBUG(log, "applyRowEvent(): " << roi.m_table_id << " " << key.first << "." << key.second);

    const auto& table = rli.getTable(key);

    if (table) {

        LOG_DEBUG(log, "Table " << table->database_name << "." << table->table_name << " has callback.");


        unsigned char* row_start = roi.m_rows_buf;

        if (should_process(table->m_filter, kind)) {
            while (row_start < roi.m_rows_end &&
                   row_start != NULL) {
                time_stamp start = now();
                try
                {
                    if (kind == eUpdate) {

                        row_start = do_update_row(*table, bei, roi, row_start, ext_state);

                    } else {
                        row_start = do_writedelete_row(*table, bei, roi, row_start, ext_state);
                    }
                }
                catch (...)
                {
                    if (event_stat)
                        event_stat->tickModifyEventFailed(roi.m_table_id, kind);
                    throw;
                }
                if (event_stat)
                    event_stat->tickModifyRowDone(roi.m_table_id, kind, now() - start);
            }

            if (event_stat)
                event_stat->tickModifyEventDone(roi.m_table_id, kind);
            return;
        }
        else if (event_stat)
            event_stat->tickModifyEventFiltered(roi.m_table_id, kind);
    }
    if (event_stat)
        event_stat->tickModifyEventIgnored(roi.m_table_id, kind);
}



} // namespace