private void parse_geometry_type(LogBuffer buffer, int length) {
    // stores geometry column's types extracted from field.
    int limit = buffer.position() + length;

    List<Integer> datas = new ArrayList<Integer>();
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int col_type = (int) buffer.getPackedLong();
        datas.add(col_type);
    }

    int index = 0;
    for (int i = 0; i < columnCnt; i++) {
        if (columnInfo[i].type == LogEvent.MYSQL_TYPE_GEOMETRY) {
            columnInfo[i].geoType = datas.get(index);
            index++;
        }
    }
}
 

public IncidentLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    final int postHeaderLen = descriptionEvent.postHeaderLen[header.type - 1];

    buffer.position(commonHeaderLen);
    final int incidentNumber = buffer.getUint16();
    if (incidentNumber >= INCIDENT_COUNT || incidentNumber <= 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().
        incident = INCIDENT_NONE;
        message = null;
        return;
    }
    incident = incidentNumber;

    buffer.position(commonHeaderLen + postHeaderLen);
    message = buffer.getString();
}
 

public XaPrepareLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.getCommonHeaderLen();
    final int postHeaderLen = descriptionEvent.getPostHeaderLen()[header.getType() - 1];

    int offset = commonHeaderLen + postHeaderLen;
    buffer.position(offset);

    onePhase = (buffer.getInt8() == 0x00 ? false : true);

    formatId = buffer.getInt32();
    gtridLength = buffer.getInt32();
    bqualLength = buffer.getInt32();

    int MY_XIDDATASIZE = 128;
    if (MY_XIDDATASIZE >= gtridLength + bqualLength && gtridLength >= 0 && gtridLength <= 64 && bqualLength >= 0
        && bqualLength <= 64) {
        data = buffer.getData(gtridLength + bqualLength);
    } else {
        formatId = -1;
        gtridLength = 0;
        bqualLength = 0;
    }
}
 

private void parse_signedness(LogBuffer buffer, int length) {
    // stores the signedness flags extracted from field
    List<Boolean> datas = new ArrayList<Boolean>();
    for (int i = 0; i < length; i++) {
        int ut = buffer.getUint8();
        for (int c = 0x80; c != 0; c >>= 1) {
            datas.add((ut & c) > 0);
        }
    }

    int index = 0;
    for (int i = 0; i < columnCnt; i++) {
        if (is_numeric_type(columnInfo[i].type)) {
            columnInfo[i].unsigned = datas.get(index);
            index++;
        }
    }
}
 

private void parse_geometry_type(LogBuffer buffer, int length) {
    // stores geometry column's types extracted from field.
    int limit = buffer.position() + length;

    List<Integer> datas = new ArrayList<Integer>();
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int col_type = (int) buffer.getPackedLong();
        datas.add(col_type);
    }

    int index = 0;
    for (int i = 0; i < columnCnt; i++) {
        if (columnInfo[i].type == LogEvent.MYSQL_TYPE_GEOMETRY) {
            columnInfo[i].geoType = datas.get(index);
            index++;
        }
    }
}
 

/**
 * Creates a new <code>Rotate_log_event</code> object read normally from
 * log.
 * 
 * @throws MySQLExtractException
 */
public RotateLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int headerSize = descriptionEvent.commonHeaderLen;
    final int postHeaderLen = descriptionEvent.postHeaderLen[ROTATE_EVENT - 1];

    buffer.position(headerSize + R_POS_OFFSET);
    position = (postHeaderLen != 0) ? buffer.getLong64() : 4; // !uint8korr(buf
                                                              // +
                                                              // R_POS_OFFSET)

    final int filenameOffset = headerSize + postHeaderLen;
    int filenameLen = buffer.limit() - filenameOffset;
    if (filenameLen > FN_REFLEN - 1) filenameLen = FN_REFLEN - 1;
    buffer.position(filenameOffset);

    filename = buffer.getFixString(filenameLen);
}
 

public XaPrepareLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.getCommonHeaderLen();
    final int postHeaderLen = descriptionEvent.getPostHeaderLen()[header.getType() - 1];

    int offset = commonHeaderLen + postHeaderLen;
    buffer.position(offset);

    onePhase = (buffer.getInt8() == 0x00 ? false : true);

    formatId = buffer.getInt32();
    gtridLength = buffer.getInt32();
    bqualLength = buffer.getInt32();

    int MY_XIDDATASIZE = 128;
    if (MY_XIDDATASIZE >= gtridLength + bqualLength && gtridLength >= 0 && gtridLength <= 64 && bqualLength >= 0
        && bqualLength <= 64) {
        data = buffer.getData(gtridLength + bqualLength);
    } else {
        formatId = -1;
        gtridLength = 0;
        bqualLength = 0;
    }
}
 

@Override
public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException {
    updateSettings();
    loadBinlogChecksum();
    sendBinlogDumpGTID(gtidSet);
    ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
    ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum);
    DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
    try {
        fetcher.start(connector.getChannel());
        while (fetcher.fetch()) {
            accumulateReceivedBytes(fetcher.limit());
            LogBuffer buffer = fetcher.duplicate();
            fetcher.consume(fetcher.limit());
            if (!coprocessor.publish(buffer)) {
                break;
            }
        }
    } finally {
        fetcher.close();
    }
}
 

public IncidentLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    final int postHeaderLen = descriptionEvent.postHeaderLen[header.type - 1];

    buffer.position(commonHeaderLen);
    final int incidentNumber = buffer.getUint16();
    if (incidentNumber >= INCIDENT_COUNT || incidentNumber <= 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().
        incident = INCIDENT_NONE;
        message = null;
        return;
    }
    incident = incidentNumber;

    buffer.position(commonHeaderLen + postHeaderLen);
    message = buffer.getString();
}
 

public ExecuteLoadQueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent)
                                                                                                               throws IOException{
    super(header, buffer, descriptionEvent);

    buffer.position(descriptionEvent.commonHeaderLen + ELQ_FILE_ID_OFFSET);

    fileId = buffer.getUint32(); // ELQ_FILE_ID_OFFSET
    fnPosStart = (int) buffer.getUint32(); // ELQ_FN_POS_START_OFFSET
    fnPosEnd = (int) buffer.getUint32(); // ELQ_FN_POS_END_OFFSET
    dupHandling = buffer.getInt8(); // ELQ_DUP_HANDLING_OFFSET

    final int len = query.length();
    if (fnPosStart > len || fnPosEnd > len || dupHandling > LOAD_DUP_REPLACE) {
        throw new IOException(String.format("Invalid ExecuteLoadQueryLogEvent: fn_pos_start=%d, "
                                            + "fn_pos_end=%d, dup_handling=%d", fnPosStart, fnPosEnd, dupHandling));
    }
}
 

private void parse_set_str_value(LogBuffer buffer, int length, boolean set) {
    // stores SET/ENUM column's string values extracted from
    // field. Each SET/ENUM column's string values are stored
    // into a string separate vector. All of them are stored
    // in 'vec'.
    int limit = buffer.position() + length;
    List<List<String>> datas = new ArrayList<List<String>>();
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int count = (int) buffer.getPackedLong();
        List<String> data = new ArrayList<String>(count);
        for (int i = 0; i < count; i++) {
            int len1 = (int) buffer.getPackedLong();
            data.add(buffer.getFixString(len1));
        }

        datas.add(data);
    }

    int index = 0;
    for (int i = 0; i < columnCnt; i++) {
        if (set && getRealType(columnInfo[i].type, columnInfo[i].meta) == LogEvent.MYSQL_TYPE_SET) {
            columnInfo[i].set_enum_values = datas.get(index);
            index++;
        }

        if (!set && getRealType(columnInfo[i].type, columnInfo[i].meta) == LogEvent.MYSQL_TYPE_ENUM) {
            columnInfo[i].set_enum_values = datas.get(index);
            index++;
        }
    }
}
 

public AppendBlockLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    final int postHeaderLen = descriptionEvent.postHeaderLen[header.type - 1];
    final int totalHeaderLen = commonHeaderLen + postHeaderLen;

    buffer.position(commonHeaderLen + AB_FILE_ID_OFFSET);
    fileId = buffer.getUint32();

    buffer.position(postHeaderLen);
    blockLen = buffer.limit() - totalHeaderLen;
    blockBuf = buffer.duplicate(blockLen);
}
 

public GtidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    // final int postHeaderLen = descriptionEvent.postHeaderLen[header.type
    // - 1];

    buffer.position(commonHeaderLen);
    commitFlag = (buffer.getUint8() != 0); // ENCODED_FLAG_LENGTH

    byte[] bs = buffer.getData(ENCODED_SID_LENGTH);
    ByteBuffer bb = ByteBuffer.wrap(bs);
    long high = bb.getLong();
    long low = bb.getLong();
    sid = new UUID(high, low);

    gno = buffer.getLong64();

    // support gtid lastCommitted and sequenceNumber
    // fix bug #776
    if (buffer.hasRemaining() && buffer.remaining() > 16 && buffer.getUint8() == LOGICAL_TIMESTAMP_TYPE_CODE) {
        lastCommitted = buffer.getLong64();
        sequenceNumber = buffer.getLong64();
    }

    // ignore gtid info read
    // sid.copy_from((uchar *)ptr_buffer);
    // ptr_buffer+= ENCODED_SID_LENGTH;
    //
    // // SIDNO is only generated if needed, in get_sidno().
    // spec.gtid.sidno= -1;
    //
    // spec.gtid.gno= uint8korr(ptr_buffer);
    // ptr_buffer+= ENCODED_GNO_LENGTH;
}
 

private void parse_simple_pk(LogBuffer buffer, int length) {
    // stores primary key's column information extracted from
    // field. Each column has an index and a prefix which are
    // stored as a unit_pair. prefix is always 0 for
    // SIMPLE_PRIMARY_KEY field.

    int limit = buffer.position() + length;
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int col_index = (int) buffer.getPackedLong();
        columnInfo[col_index].pk = true;
    }
}
 

public ExecuteLoadLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    buffer.position(commonHeaderLen + EL_FILE_ID_OFFSET);
    fileId = buffer.getUint32(); // EL_FILE_ID_OFFSET
}
 

private void parse_pk_with_prefix(LogBuffer buffer, int length) {
    // stores primary key's column information extracted from
    // field. Each column has an index and a prefix which are
    // stored as a unit_pair.
    int limit = buffer.position() + length;
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int col_index = (int) buffer.getPackedLong();
        // prefix length, 比如 char(32)
        @SuppressWarnings("unused")
        int col_prefix = (int) buffer.getPackedLong();
        columnInfo[col_index].pk = true;
    }
}
 

public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName, int jsonColumnCount, boolean partial){
    this.buffer = buffer;
    this.columnLen = columnLen;
    this.charsetName = charsetName;
    this.partial = partial;
    this.jsonColumnCount = jsonColumnCount;
    this.nullBits = new BitSet(columnLen);
    this.partialBits = new BitSet(1);
}
 

private List<Integer> parse_column_charset(LogBuffer buffer, int length) {
    // stores collation numbers extracted from field.
    int limit = buffer.position() + length;
    List<Integer> datas = new ArrayList<Integer>();
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int col_charset = (int) buffer.getPackedLong();
        datas.add(col_charset);
    }

    return datas;
}
 

private void parse_column_name(LogBuffer buffer, int length) {
    // stores column names extracted from field
    int limit = buffer.position() + length;
    int index = 0;
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int len = (int) buffer.getPackedLong();
        columnInfo[index++].name = buffer.getFixString(len);
    }
}
 

public XidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    /* The Post-Header is empty. The Variable Data part begins immediately. */
    buffer.position(descriptionEvent.commonHeaderLen + descriptionEvent.postHeaderLen[XID_EVENT - 1]);
    xid = buffer.getLong64(); // !uint8korr
}
 

public RowsQueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header, buffer, descriptionEvent);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    final int postHeaderLen = descriptionEvent.postHeaderLen[header.type - 1];

    /*
     * m_rows_query length is stored using only one byte, but that length is
     * ignored and the complete query is read.
     */
    int offset = commonHeaderLen + postHeaderLen + 1;
    int len = buffer.limit() - offset;
    rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
}
 

private void parse_pk_with_prefix(LogBuffer buffer, int length) {
    // stores primary key's column information extracted from
    // field. Each column has an index and a prefix which are
    // stored as a unit_pair.
    int limit = buffer.position() + length;
    while (buffer.hasRemaining() && buffer.position() < limit) {
        int col_index = (int) buffer.getPackedLong();
        // prefix length, 比如 char(32)
        @SuppressWarnings("unused")
        int col_prefix = (int) buffer.getPackedLong();
        columnInfo[col_index].pk = true;
    }
}
 

public IntvarLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    /* The Post-Header is empty. The Varible Data part begins immediately. */
    buffer.position(descriptionEvent.commonHeaderLen + descriptionEvent.postHeaderLen[INTVAR_EVENT - 1]
                    + I_TYPE_OFFSET);
    type = buffer.getInt8(); // I_TYPE_OFFSET
    value = buffer.getLong64(); // !uint8korr(buf + I_VAL_OFFSET);
}
 

public DeleteFileLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    buffer.position(commonHeaderLen + DF_FILE_ID_OFFSET);
    fileId = buffer.getUint32(); // DF_FILE_ID_OFFSET
}
 

public XidLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    /* The Post-Header is empty. The Variable Data part begins immediately. */
    buffer.position(descriptionEvent.commonHeaderLen + descriptionEvent.postHeaderLen[XID_EVENT - 1]);
    xid = buffer.getLong64(); // !uint8korr
}
 

public StartLogEventV3(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    buffer.position(descriptionEvent.commonHeaderLen);
    binlogVersion = buffer.getUint16(); // ST_BINLOG_VER_OFFSET
    serverVersion = buffer.getFixString(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
}
 

public DeleteFileLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    buffer.position(commonHeaderLen + DF_FILE_ID_OFFSET);
    fileId = buffer.getUint32(); // DF_FILE_ID_OFFSET
}
 

public RowsQueryLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header, buffer, descriptionEvent);

    final int commonHeaderLen = descriptionEvent.commonHeaderLen;
    final int postHeaderLen = descriptionEvent.postHeaderLen[header.type - 1];

    /*
     * m_rows_query length is stored using only one byte, but that length is
     * ignored and the complete query is read.m_rows_query长度只使用一个字节存储，但是这个长度是
     *被忽略，然后读取完整的查询。
     */
    int offset = commonHeaderLen + postHeaderLen + 1;
    int len = buffer.limit() - offset;
    rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
}
 

public AnnotateRowsEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
    super(header, buffer, descriptionEvent);

    final int commonHeaderLen = descriptionEvent.getCommonHeaderLen();
    final int postHeaderLen = descriptionEvent.getPostHeaderLen()[header.getType() - 1];

    int offset = commonHeaderLen + postHeaderLen;
    int len = buffer.limit() - offset;
    rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
}
 

public FormatDescriptionLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent)
                                                                                                                throws IOException{
    /* Start_log_event_v3 */
    super(header, buffer, descriptionEvent);

    buffer.position(LOG_EVENT_MINIMAL_HEADER_LEN + ST_COMMON_HEADER_LEN_OFFSET);
    commonHeaderLen = buffer.getUint8();
    if (commonHeaderLen < OLD_HEADER_LEN) /* sanity check */
    {
        throw new IOException("Format Description event header length is too short");
    }

    numberOfEventTypes = buffer.limit() - (LOG_EVENT_MINIMAL_HEADER_LEN + ST_COMMON_HEADER_LEN_OFFSET + 1);

    // buffer.position(LOG_EVENT_MINIMAL_HEADER_LEN
    // + ST_COMMON_HEADER_LEN_OFFSET + 1);
    postHeaderLen = new short[numberOfEventTypes];
    for (int i = 0; i < numberOfEventTypes; i++) {
        postHeaderLen[i] = (short) buffer.getUint8();
    }

    calcServerVersionSplit();
    long calc = getVersionProduct();
    if (calc >= checksumVersionProduct) {
        /*
         * the last bytes are the checksum alg desc and value (or value's
         * room)
         */
        numberOfEventTypes -= BINLOG_CHECKSUM_ALG_DESC_LEN;
    }

    if (logger.isInfoEnabled()) logger.info("common_header_len= " + commonHeaderLen + ", number_of_event_types= "
                                            + numberOfEventTypes);
}