content: package com.alibaba.otter.canal.parse.inbound.mysql.dbsync;
  
  import java.io.Serializable;
  import java.io.UnsupportedEncodingException;
  import java.math.BigDecimal;
  import java.math.BigInteger;
  import java.nio.charset.Charset;
  import java.sql.Types;
  import java.util.Arrays;
  import java.util.BitSet;
  import java.util.HashMap;
  import java.util.List;
  import java.util.Map;
  
  import com.taobao.tddl.dbsync.binlog.event.*;
  import org.apache.commons.lang.StringUtils;
  import org.apache.commons.lang.exception.ExceptionUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.alibaba.otter.canal.common.AbstractCanalLifeCycle;
  import com.alibaba.otter.canal.filter.aviater.AviaterRegexFilter;
  import com.alibaba.otter.canal.parse.exception.CanalParseException;
  import com.alibaba.otter.canal.parse.inbound.BinlogParser;
  import com.alibaba.otter.canal.parse.inbound.TableMeta;
  import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
  import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DdlResult;
  import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
  import com.alibaba.otter.canal.parse.inbound.mysql.ddl.SimpleDdlParser;
  import com.alibaba.otter.canal.protocol.CanalEntry.Column;
  import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
  import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
  import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
  import com.alibaba.otter.canal.protocol.CanalEntry.Header;
  import com.alibaba.otter.canal.protocol.CanalEntry.Pair;
  import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
  import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
  import com.alibaba.otter.canal.protocol.CanalEntry.TransactionBegin;
  import com.alibaba.otter.canal.protocol.CanalEntry.TransactionEnd;
  import com.alibaba.otter.canal.protocol.CanalEntry.Type;
  import com.alibaba.otter.canal.protocol.position.EntryPosition;
  import com.google.protobuf.ByteString;
  import com.taobao.tddl.dbsync.binlog.LogEvent;
  import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
  import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
  import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidListLogEvent;
  import com.taobao.tddl.dbsync.binlog.event.mariadb.MariaGtidLogEvent;
  import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException;
  
  /**
   * 基于{@linkplain LogEvent}转化为Entry对象的处理
   * 
   * @author jianghang 2013-1-17 下午02:41:14
   * @version 1.0.0
   */
  public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogParser<LogEvent> {
  
      public static final String          XA_XID              = "XA_XID";
      public static final String          XA_TYPE             = "XA_TYPE";
      public static final String          XA_START            = "XA START";
      public static final String          XA_END              = "XA END";
      public static final String          XA_COMMIT           = "XA COMMIT";
      public static final String          XA_ROLLBACK         = "XA ROLLBACK";
      public static final String          ISO_8859_1          = "ISO-8859-1";
      public static final String          UTF_8               = "UTF-8";
      public static final int             TINYINT_MAX_VALUE   = 256;
      public static final int             SMALLINT_MAX_VALUE  = 65536;
      public static final int             MEDIUMINT_MAX_VALUE = 16777216;
      public static final long            INTEGER_MAX_VALUE   = 4294967296L;
      public static final BigInteger      BIGINT_MAX_VALUE    = new BigInteger("18446744073709551616");
      public static final int             version             = 1;
      public static final String          BEGIN               = "BEGIN";
      public static final String          COMMIT              = "COMMIT";
      public static final Logger          logger              = LoggerFactory.getLogger(LogEventConvert.class);
  
      private volatile AviaterRegexFilter nameFilter;                                                          // 运行时引用可能会有变化，比如规则发生变化时
      private volatile AviaterRegexFilter nameBlackFilter;
      private Map<String, List<String>>   fieldFilterMap      = new HashMap<>();
      private Map<String, List<String>>   fieldBlackFilterMap = new HashMap<>();
  
      private TableMetaCache              tableMetaCache;
      private Charset                     charset             = Charset.defaultCharset();
      private boolean                     filterQueryDcl      = false;
      private boolean                     filterQueryDml      = false;
      private boolean                     filterQueryDdl      = false;
      // 是否跳过table相关的解析异常,比如表不存在或者列数量不匹配,issue 92
      private boolean                     filterTableError    = false;
      // 新增rows过滤，用于仅订阅除rows以外的数据
      private boolean                     filterRows          = false;
      private boolean                     useDruidDdlFilter   = true;
  
      public LogEventConvert(){
  
      }
  
      @Override
      public Entry parse(LogEvent logEvent, boolean isSeek) throws CanalParseException {
          if (logEvent == null || logEvent instanceof UnknownLogEvent) {
              return null;
          }
  
          int eventType = logEvent.getHeader().getType();
          switch (eventType) {
              case LogEvent.QUERY_EVENT:
                  return parseQueryEvent((QueryLogEvent) logEvent, isSeek);
              case LogEvent.XID_EVENT:
                  return parseXidEvent((XidLogEvent) logEvent);
              case LogEvent.TABLE_MAP_EVENT:
                  parseTableMapEvent((TableMapLogEvent) logEvent);
                  break;
              case LogEvent.WRITE_ROWS_EVENT_V1:
              case LogEvent.WRITE_ROWS_EVENT:
                  return parseRowsEvent((WriteRowsLogEvent) logEvent);
              case LogEvent.UPDATE_ROWS_EVENT_V1:
              case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
              case LogEvent.UPDATE_ROWS_EVENT:
                  return parseRowsEvent((UpdateRowsLogEvent) logEvent);
              case LogEvent.DELETE_ROWS_EVENT_V1:
              case LogEvent.DELETE_ROWS_EVENT:
                  return parseRowsEvent((DeleteRowsLogEvent) logEvent);
              case LogEvent.ROWS_QUERY_LOG_EVENT:
                  return parseRowsQueryEvent((RowsQueryLogEvent) logEvent);
              case LogEvent.ANNOTATE_ROWS_EVENT:
                  return parseAnnotateRowsEvent((AnnotateRowsEvent) logEvent);
              case LogEvent.USER_VAR_EVENT:
                  return parseUserVarLogEvent((UserVarLogEvent) logEvent);
              case LogEvent.INTVAR_EVENT:
                  return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
              case LogEvent.RAND_EVENT:
                  return parseRandLogEvent((RandLogEvent) logEvent);
              case LogEvent.GTID_LOG_EVENT:
                  return parseGTIDLogEvent((GtidLogEvent) logEvent);
              case LogEvent.HEARTBEAT_LOG_EVENT:
                  return parseHeartbeatLogEvent((HeartbeatLogEvent) logEvent);
              case LogEvent.HEARTBEAT_LOG_EVENT_V2:
                  return parseHeartbeatV2LogEvent((HeartbeatV2LogEvent) logEvent);
              case LogEvent.GTID_EVENT:
              case LogEvent.GTID_LIST_EVENT:
                  return parseMariaGTIDLogEvent(logEvent);
              default:
                  break;
          }
  
          return null;
      }
  
      public void reset() {
          // do nothing
          if (tableMetaCache != null) {
              tableMetaCache.clearTableMeta();
          }
      }
  
      private Entry parseHeartbeatLogEvent(HeartbeatLogEvent logEvent) {
          Header.Builder headerBuilder = Header.newBuilder();
          headerBuilder.setEventType(EventType.MHEARTBEAT);
          Entry.Builder entryBuilder = Entry.newBuilder();
          entryBuilder.setHeader(headerBuilder.build());
          entryBuilder.setEntryType(EntryType.HEARTBEAT);
          return entryBuilder.build();
      }
  
      private Entry parseHeartbeatV2LogEvent(HeartbeatV2LogEvent logEvent) {
          Header.Builder headerBuilder = Header.newBuilder();
          headerBuilder.setEventType(EventType.MHEARTBEAT);
          Entry.Builder entryBuilder = Entry.newBuilder();
          entryBuilder.setHeader(headerBuilder.build());
          entryBuilder.setEntryType(EntryType.HEARTBEAT);
          return entryBuilder.build();
      }
  
      private Entry parseGTIDLogEvent(GtidLogEvent logEvent) {
          LogHeader logHeader = logEvent.getHeader();
          Pair.Builder builder = Pair.newBuilder();
          builder.setKey("gtid");
          builder.setValue(logEvent.getGtidStr());
  
          if (logEvent.getLastCommitted() != -1) {
              builder.setKey("lastCommitted");
              builder.setValue(String.valueOf(logEvent.getLastCommitted()));
              builder.setKey("sequenceNumber");
              builder.setValue(String.valueOf(logEvent.getSequenceNumber()));
          }
  
          Header header = createHeader(logHeader, "", "", EventType.GTID);
          return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
      }
  
      private Entry parseMariaGTIDLogEvent(LogEvent logEvent) {
          LogHeader logHeader = logEvent.getHeader();
          Pair.Builder builder = Pair.newBuilder();
          builder.setKey("gtid");
          if (logEvent instanceof MariaGtidLogEvent) {
              builder.setValue(((MariaGtidLogEvent) logEvent).getGtidStr());
          } else if (logEvent instanceof MariaGtidListLogEvent) {
              builder.setValue(((MariaGtidListLogEvent) logEvent).getGtidStr());
          }
          Header header = createHeader(logHeader, "", "", EventType.GTID);
          return createEntry(header, EntryType.GTIDLOG, builder.build().toByteString());
      }
  
      private Entry parseQueryEvent(QueryLogEvent event, boolean isSeek) {
          String queryString = event.getQuery();
          if (StringUtils.startsWithIgnoreCase(queryString, XA_START)) {
              // xa start use TransactionBegin
              TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
              beginBuilder.setThreadId(event.getSessionId());
              beginBuilder.addProps(createSpecialPair(XA_TYPE, XA_START));
              beginBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_START)));
              TransactionBegin transactionBegin = beginBuilder.build();
              Header header = createHeader(event.getHeader(), "", "", null);
              return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
          } else if (StringUtils.startsWithIgnoreCase(queryString, XA_END)) {
              // xa start use TransactionEnd
              TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
              endBuilder.setTransactionId(String.valueOf(0L));
              endBuilder.addProps(createSpecialPair(XA_TYPE, XA_END));
              endBuilder.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_END)));
              TransactionEnd transactionEnd = endBuilder.build();
              Header header = createHeader(event.getHeader(), "", "", null);
              return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
          } else if (StringUtils.startsWithIgnoreCase(queryString, XA_COMMIT)) {
              // xa commit
              Header header = createHeader(event.getHeader(), "", "", EventType.XACOMMIT);
              RowChange.Builder rowChangeBuider = RowChange.newBuilder();
              rowChangeBuider.setSql(queryString);
              rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_COMMIT));
              rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_COMMIT)));
              rowChangeBuider.setEventType(EventType.XACOMMIT);
              return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
          } else if (StringUtils.startsWithIgnoreCase(queryString, XA_ROLLBACK)) {
              // xa rollback
              Header header = createHeader(event.getHeader(), "", "", EventType.XAROLLBACK);
              RowChange.Builder rowChangeBuider = RowChange.newBuilder();
              rowChangeBuider.setSql(queryString);
              rowChangeBuider.addProps(createSpecialPair(XA_TYPE, XA_ROLLBACK));
              rowChangeBuider.addProps(createSpecialPair(XA_XID, getXaXid(queryString, XA_ROLLBACK)));
              rowChangeBuider.setEventType(EventType.XAROLLBACK);
              return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
          } else if (StringUtils.endsWithIgnoreCase(queryString, BEGIN)) {
              TransactionBegin transactionBegin = createTransactionBegin(event.getSessionId());
              Header header = createHeader(event.getHeader(), "", "", null);
              return createEntry(header, EntryType.TRANSACTIONBEGIN, transactionBegin.toByteString());
          } else if (StringUtils.endsWithIgnoreCase(queryString, COMMIT)) {
              TransactionEnd transactionEnd = createTransactionEnd(0L); // MyISAM可能不会有xid事件
              Header header = createHeader(event.getHeader(), "", "", null);
              return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
          } else {
              boolean notFilter = false;
              EventType type = EventType.QUERY;
              String tableName = null;
              String schemaName = null;
              if (useDruidDdlFilter) {
                  List<DdlResult> results = DruidDdlParser.parse(queryString, event.getDbName());
                  for (DdlResult result : results) {
                      if (!processFilter(queryString, result)) {
                          // 只要有一个数据不进行过滤
                          notFilter = true;
                      }
                  }
                  if (results.size() > 0) {
                      // 如果针对多行的DDL,只能取第一条
                      type = results.get(0).getType();
                      schemaName = results.get(0).getSchemaName();
                      tableName = results.get(0).getTableName();
                  }
              } else {
                  DdlResult result = SimpleDdlParser.parse(queryString, event.getDbName());
                  if (!processFilter(queryString, result)) {
                      notFilter = true;
                  }
  
                  type = result.getType();
                  schemaName = result.getSchemaName();
                  tableName = result.getTableName();
              }
  
              if (!notFilter) {
                  // 如果是过滤的数据就不处理了
                  return null;
              }
  
              boolean isDml = (type == EventType.INSERT || type == EventType.UPDATE || type == EventType.DELETE);
  
              // filterQueryDdl=true的情况下,也得更新tablemeta
              if (!isSeek && !isDml) {
                  // 使用新的表结构元数据管理方式
                  EntryPosition position = createPosition(event.getHeader());
                  tableMetaCache.apply(position, event.getDbName(), queryString, null);
              }
  
              if (filterQueryDdl) {
                  // 全部DDL过滤,那就忽略事件生成
                  return null;
              }
  
              Header header = createHeader(event.getHeader(), schemaName, tableName, type);
              RowChange.Builder rowChangeBuilder = RowChange.newBuilder();
              rowChangeBuilder.setIsDdl(!isDml);
              rowChangeBuilder.setSql(queryString);
              if (StringUtils.isNotEmpty(event.getDbName())) {// 可能为空
                  rowChangeBuilder.setDdlSchemaName(event.getDbName());
              }
              rowChangeBuilder.setEventType(type);
              return createEntry(header, EntryType.ROWDATA, rowChangeBuilder.build().toByteString());
          }
      }
  
      private String getXaXid(String queryString, String type) {
          return StringUtils.substringAfter(queryString, type);
      }
  
      private boolean processFilter(String queryString, DdlResult result) {
          String schemaName = result.getSchemaName();
          String tableName = result.getTableName();
          // fixed issue https://github.com/alibaba/canal/issues/58
          // 更新下table meta cache
          if (tableMetaCache != null
              && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) {
              // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
              for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) {
                  String schemaName0 = renameResult.getSchemaName();
                  String tableName0 = renameResult.getTableName();
                  if (StringUtils.isNotEmpty(tableName0)) {
                      // 如果解析到了正确的表信息，则根据全名进行清除
                      tableMetaCache.clearTableMeta(schemaName0, tableName0);
                  } else {
                      // 如果无法解析正确的表信息，则根据schema进行清除
                      tableMetaCache.clearTableMetaWithSchemaName(schemaName0);
                  }
              }
          }
  
          // fixed issue https://github.com/alibaba/canal/issues/58
          if (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE
              || result.getType() == EventType.CREATE || result.getType() == EventType.TRUNCATE
              || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX
              || result.getType() == EventType.DINDEX) { // 针对DDL类型
  
              if (!filterQueryDdl && (StringUtils.isEmpty(tableName)
                  || (result.getType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName())))) {
                  // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析
                  throw new CanalParseException("SimpleDdlParser process query failed. pls submit issue with this queryString: "
                                                + queryString + " , and DdlResult: " + result.toString());
                  // return null;
              } else {
                  // check name filter
                  String name = schemaName + "." + tableName;
                  if (nameFilter != null && !nameFilter.filter(name)) {
                      if (result.getType() == EventType.RENAME) {
                          // rename校验只要源和目标满足一个就进行操作
                          if (nameFilter != null
                              && !nameFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) {
                              return true;
                          }
                      } else {
                          // 其他情况返回null
                          return true;
                      }
                  }
  
                  if (nameBlackFilter != null && nameBlackFilter.filter(name)) {
                      if (result.getType() == EventType.RENAME) {
                          // rename校验只要源和目标满足一个就进行操作
                          if (nameBlackFilter != null
                              && nameBlackFilter.filter(result.getOriSchemaName() + "." + result.getOriTableName())) {
                              return true;
                          }
                      } else {
                          // 其他情况返回null
                          return true;
                      }
                  }
              }
          } else if (result.getType() == EventType.INSERT || result.getType() == EventType.UPDATE
                     || result.getType() == EventType.DELETE) {
              // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤
              if (filterQueryDml) {
                  return true;
              }
          } else if (filterQueryDcl) {
              return true;
          }
  
          return false;
      }
  
      private Entry parseRowsQueryEvent(RowsQueryLogEvent event) {
          if (filterQueryDml) {
              return null;
          }
          // mysql5.6支持，需要设置binlog-rows-query-log-events=1，可详细打印原始DML语句
          String queryString = null;
          try {
              queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset);
              String tableName = null;
              if (useDruidDdlFilter) {
                  List<DdlResult> results = DruidDdlParser.parse(queryString, null);
                  if (results.size() > 0) {
                      tableName = results.get(0).getTableName();
                  }
              }
  
              return buildQueryEntry(queryString, event.getHeader(), tableName);
          } catch (UnsupportedEncodingException e) {
              throw new CanalParseException(e);
          }
      }
  
      private Entry parseAnnotateRowsEvent(AnnotateRowsEvent event) {
          if (filterQueryDml) {
              return null;
          }
          // mariaDb支持，需要设置binlog_annotate_row_events=true，可详细打印原始DML语句
          String queryString = null;
          try {
              queryString = new String(event.getRowsQuery().getBytes(ISO_8859_1), charset);
              return buildQueryEntry(queryString, event.getHeader());
          } catch (UnsupportedEncodingException e) {
              throw new CanalParseException(e);
          }
      }
  
      private Entry parseUserVarLogEvent(UserVarLogEvent event) {
          if (filterQueryDml) {
              return null;
          }
  
          return buildQueryEntry(event.getQuery(), event.getHeader());
      }
  
      private Entry parseIntrvarLogEvent(IntvarLogEvent event) {
          if (filterQueryDml) {
              return null;
          }
  
          return buildQueryEntry(event.getQuery(), event.getHeader());
      }
  
      private Entry parseRandLogEvent(RandLogEvent event) {
          if (filterQueryDml) {
              return null;
          }
  
          return buildQueryEntry(event.getQuery(), event.getHeader());
      }
  
      private Entry parseXidEvent(XidLogEvent event) {
          TransactionEnd transactionEnd = createTransactionEnd(event.getXid());
          Header header = createHeader(event.getHeader(), "", "", null);
          return createEntry(header, EntryType.TRANSACTIONEND, transactionEnd.toByteString());
      }
  
      public TableMeta parseRowsEventForTableMeta(RowsLogEvent event) {
          TableMapLogEvent table = event.getTable();
          if (table == null) {
              // tableId对应的记录不存在
              throw new TableIdNotFoundException("not found tableId:" + event.getTableId());
          }
  
          boolean isHeartBeat = isAliSQLHeartBeat(table.getDbName(), table.getTableName());
          boolean isRDSHeartBeat = tableMetaCache.isOnRDS() && isRDSHeartBeat(table.getDbName(), table.getTableName());
  
          String fullname = table.getDbName() + "." + table.getTableName();
          // check name filter
          if (nameFilter != null && !nameFilter.filter(fullname)) {
              return null;
          }
          if (nameBlackFilter != null && nameBlackFilter.filter(fullname)) {
              return null;
          }
  
          // if (isHeartBeat || isRDSHeartBeat) {
          // // 忽略rds模式的mysql.ha_health_check心跳数据
          // return null;
          // }
          TableMeta tableMeta = null;
          if (isRDSHeartBeat) {
              // 处理rds模式的mysql.ha_health_check心跳数据
              // 主要RDS的心跳表基本无权限,需要mock一个tableMeta
              FieldMeta idMeta = new FieldMeta("id", "bigint(20)", true, false, "0");
              FieldMeta typeMeta = new FieldMeta("type", "char(1)", false, true, "0");
              tableMeta = new TableMeta(table.getDbName(), table.getTableName(), Arrays.asList(idMeta, typeMeta));
          } else if (isHeartBeat) {
              // 处理alisql模式的test.heartbeat心跳数据
              // 心跳表基本无权限,需要mock一个tableMeta
              FieldMeta idMeta = new FieldMeta("id", "smallint(6)", false, true, null);
              FieldMeta typeMeta = new FieldMeta("ts", "int(11)", true, false, null);
              tableMeta = new TableMeta(table.getDbName(), table.getTableName(), Arrays.asList(idMeta, typeMeta));
          }
  
          EntryPosition position = createPosition(event.getHeader());
          if (tableMetaCache != null && tableMeta == null) {// 入错存在table meta
              tableMeta = getTableMeta(table.getDbName(), table.getTableName(), true, position);
              if (tableMeta == null) {
                  if (!filterTableError) {
                      throw new CanalParseException("not found [" + fullname + "] in db , pls check!");
                  }
              }
          }
  
          return tableMeta;
      }
  
      public Entry parseRowsEvent(RowsLogEvent event) {
          return parseRowsEvent(event, null);
      }
  
      public void parseTableMapEvent(TableMapLogEvent event) {
          try {
              String charsetDbName = new String(event.getDbName().getBytes(ISO_8859_1), charset);
              event.setDbname(charsetDbName);
  
              String charsetTbName = new String(event.getTableName().getBytes(ISO_8859_1), charset);
              event.setTblname(charsetTbName);
          } catch (UnsupportedEncodingException e) {
              throw new CanalParseException(e);
          }
      }
  
      public Entry parseRowsEvent(RowsLogEvent event, TableMeta tableMeta) {
          if (filterRows) {
              return null;
          }
          try {
              if (tableMeta == null) { // 如果没有外部指定
                  tableMeta = parseRowsEventForTableMeta(event);
              }
  
              if (tableMeta == null) {
                  // 拿不到表结构,执行忽略
                  return null;
              }
  
              EventType eventType = null;
              int type = event.getHeader().getType();
              if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
                  eventType = EventType.INSERT;
              } else if (LogEvent.UPDATE_ROWS_EVENT_V1 == type || LogEvent.UPDATE_ROWS_EVENT == type
                         || LogEvent.PARTIAL_UPDATE_ROWS_EVENT == type) {
                  eventType = EventType.UPDATE;
              } else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
                  eventType = EventType.DELETE;
              } else {
                  throw new CanalParseException("unsupport event type :" + event.getHeader().getType());
              }
  
              RowChange.Builder rowChangeBuider = RowChange.newBuilder();
              rowChangeBuider.setTableId(event.getTableId());
              rowChangeBuider.setIsDdl(false);
  
              rowChangeBuider.setEventType(eventType);
              RowsLogBuffer buffer = event.getRowsBuf(charset);
              BitSet columns = event.getColumns();
              BitSet changeColumns = event.getChangeColumns();
  
              boolean tableError = false;
              int rowsCount = 0;
              while (buffer.nextOneRow(columns, false)) {
                  // 处理row记录
                  RowData.Builder rowDataBuilder = RowData.newBuilder();
                  if (EventType.INSERT == eventType) {
                      // insert的记录放在before字段中
                      tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, true, tableMeta);
                  } else if (EventType.DELETE == eventType) {
                      // delete的记录放在before字段中
                      tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta);
                  } else {
                      // update需要处理before/after
                      tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta);
                      if (!buffer.nextOneRow(changeColumns, true)) {
                          rowChangeBuider.addRowDatas(rowDataBuilder.build());
                          break;
                      }
  
                      tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta);
                  }
  
                  rowsCount++;
                  rowChangeBuider.addRowDatas(rowDataBuilder.build());
              }
              TableMapLogEvent table = event.getTable();
              Header header = createHeader(event.getHeader(),
                  table.getDbName(),
                  table.getTableName(),
                  eventType,
                  rowsCount);
  
              RowChange rowChange = rowChangeBuider.build();
              if (tableError) {
                  Entry entry = createEntry(header, EntryType.ROWDATA, ByteString.EMPTY);
                  logger.warn("table parser error : {}storeValue: {}", entry.toString(), rowChange.toString());
                  return null;
              } else {
                  Entry entry = createEntry(header, EntryType.ROWDATA, rowChange.toByteString());
                  return entry;
              }
          } catch (Exception e) {
              throw new CanalParseException("parse row data failed.", e);
          }
      }
  
      private EntryPosition createPosition(LogHeader logHeader) {
          return new EntryPosition(logHeader.getLogFileName(), logHeader.getLogPos() - logHeader.getEventLen(), // startPos
              logHeader.getWhen() * 1000L,
              logHeader.getServerId()); // 记录到秒
      }
  
      private boolean parseOneRow(RowData.Builder rowDataBuilder, RowsLogEvent event, RowsLogBuffer buffer, BitSet cols,
                                  boolean isAfter, TableMeta tableMeta) throws UnsupportedEncodingException {
          int columnCnt = event.getTable().getColumnCnt();
          ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
          // mysql8.0针对set @@global.binlog_row_metadata='FULL' 可以记录部分的metadata信息
          boolean existOptionalMetaData = event.getTable().isExistOptionalMetaData();
          boolean tableError = false;
          // check table fileds count，只能处理加字段
          boolean existRDSNoPrimaryKey = false;
          // 获取字段过滤条件
          List<String> fieldList = null;
          List<String> blackFieldList = null;
  
          if (tableMeta != null) {
              fieldList = fieldFilterMap.get(tableMeta.getFullName().toUpperCase());
              blackFieldList = fieldBlackFilterMap.get(tableMeta.getFullName().toUpperCase());
          }
  
          if (tableMeta != null && columnInfo.length > tableMeta.getFields().size()) {
              if (tableMetaCache.isOnRDS() || tableMetaCache.isOnPolarX()) {
                  // 特殊处理下RDS的场景
                  List<FieldMeta> primaryKeys = tableMeta.getPrimaryFields();
                  if (primaryKeys == null || primaryKeys.isEmpty()) {
                      if (columnInfo.length == tableMeta.getFields().size() + 1
                          && columnInfo[columnInfo.length - 1].type == LogEvent.MYSQL_TYPE_LONGLONG) {
                          existRDSNoPrimaryKey = true;
                      }
                  }
              }
  
              EntryPosition position = createPosition(event.getHeader());
              if (!existRDSNoPrimaryKey) {
                  // online ddl增加字段操作步骤：
                  // 1. 新增一张临时表，将需要做ddl表的数据全量导入
                  // 2. 在老表上建立I/U/D的trigger，增量的将数据插入到临时表
                  // 3. 锁住应用请求，将临时表rename为老表的名字，完成增加字段的操作
                  // 尝试做一次reload，可能因为ddl没有正确解析，或者使用了类似online ddl的操作
                  // 因为online ddl没有对应表名的alter语法，所以不会有clear cache的操作
                  tableMeta = getTableMeta(event.getTable().getDbName(), event.getTable().getTableName(), false, position);// 强制重新获取一次
                  if (tableMeta == null) {
                      tableError = true;
                      if (!filterTableError) {
                          throw new CanalParseException("not found [" + event.getTable().getDbName() + "."
                                                        + event.getTable().getTableName() + "] in db , pls check!");
                      }
                  }
  
                  // 在做一次判断
                  if (tableMeta != null && columnInfo.length > tableMeta.getFields().size()) {
                      tableError = true;
                      if (!filterTableError) {
                          throw new CanalParseException("column size is not match for table:" + tableMeta.getFullName()
                                                        + "," + columnInfo.length + " vs " + tableMeta.getFields().size());
                      }
                  }
                  // } else {
                  // logger.warn("[" + event.getTable().getDbName() + "." +
                  // event.getTable().getTableName()
                  // + "] is no primary key , skip alibaba_rds_row_id column");
              }
          }
  
          for (int i = 0; i < columnCnt; i++) {
              ColumnInfo info = columnInfo[i];
              // mysql 5.6开始支持nolob/mininal类型,并不一定记录所有的列,需要进行判断
              if (!cols.get(i)) {
                  continue;
              }
  
              if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
                  // 不解析最后一列
                  String rdsRowIdColumnName = "__#alibaba_rds_row_id#__";
                  if (tableMetaCache.isOnPolarX()) {
                      rdsRowIdColumnName = "_drds_implicit_id_";
                  }
                  buffer.nextValue(rdsRowIdColumnName, i, info.type, info.meta, false);
                  Column.Builder columnBuilder = Column.newBuilder();
                  columnBuilder.setName(rdsRowIdColumnName);
                  columnBuilder.setIsKey(true);
                  columnBuilder.setMysqlType("bigint");
                  columnBuilder.setIndex(i);
                  columnBuilder.setIsNull(false);
                  Serializable value = buffer.getValue();
                  columnBuilder.setValue(value.toString());
                  columnBuilder.setSqlType(Types.BIGINT);
                  columnBuilder.setUpdated(false);
  
                  if (needField(fieldList, blackFieldList, columnBuilder.getName())) {
                      if (isAfter) {
                          rowDataBuilder.addAfterColumns(columnBuilder.build());
                      } else {
                          rowDataBuilder.addBeforeColumns(columnBuilder.build());
                      }
                  }
                  continue;
              }
  
              FieldMeta fieldMeta = null;
              if (tableMeta != null && !tableError) {
                  // 处理file meta
                  fieldMeta = tableMeta.getFields().get(i);
              }
  
              if (fieldMeta != null && existOptionalMetaData && tableMetaCache.isOnTSDB()) {
                  // check column info
                  boolean check = StringUtils.equalsIgnoreCase(fieldMeta.getColumnName(), info.name);
                  check &= (fieldMeta.isUnsigned() == info.unsigned);
                  check &= (fieldMeta.isNullable() == info.nullable);
  
                  if (!check) {
                      throw new CanalParseException("MySQL8.0 unmatch column metadata & pls submit issue , table : "
                                                    + tableMeta.getFullName() + ", db fieldMeta : "
                                                    + fieldMeta.toString() + " , binlog fieldMeta : " + info.toString()
                                                    + " , on : " + event.getHeader().getLogFileName() + ":"
                                                    + (event.getHeader().getLogPos() - event.getHeader().getEventLen()));
                  }
              }
  
              Column.Builder columnBuilder = Column.newBuilder();
              if (fieldMeta != null) {
                  columnBuilder.setName(fieldMeta.getColumnName());
                  columnBuilder.setIsKey(fieldMeta.isKey());
                  // 增加mysql type类型,issue 73
                  columnBuilder.setMysqlType(fieldMeta.getColumnType());
              } else if (existOptionalMetaData) {
                  columnBuilder.setName(info.name);
                  columnBuilder.setIsKey(info.pk);
                  // mysql8.0里没有mysql type类型
                  // columnBuilder.setMysqlType(fieldMeta.getColumnType());
              }
              columnBuilder.setIndex(i);
              columnBuilder.setIsNull(false);
  
              // fixed issue
              // https://github.com/alibaba/canal/issues/66，特殊处理binary/varbinary，不能做编码处理
              boolean isBinary = false;
              if (fieldMeta != null) {
                  if (StringUtils.containsIgnoreCase(fieldMeta.getColumnType(), "VARBINARY")) {
                      isBinary = true;
                  } else if (StringUtils.containsIgnoreCase(fieldMeta.getColumnType(), "BINARY")) {
                      isBinary = true;
                  }
              }
  
              buffer.nextValue(columnBuilder.getName(), i, info.type, info.meta, isBinary);
              int javaType = buffer.getJavaType();
              if (buffer.isNull()) {
                  columnBuilder.setIsNull(true);
                  
                  // 处理各种类型
                  switch (javaType) {
                      case Types.BINARY:
                      case Types.VARBINARY:
                      case Types.LONGVARBINARY:
  
                          // https://github.com/alibaba/canal/issues/4652
                          // mysql binlog中blob/text都处理为blob类型，需要反查table
                          // meta，按编码解析text
                          if (fieldMeta != null && isText(fieldMeta.getColumnType())) {
                              javaType = Types.CLOB;
                          } else {
                              javaType = Types.BLOB;
                          }
                          break;
                  }
              } else {
                  final Serializable value = buffer.getValue();
                  // 处理各种类型
                  switch (javaType) {
                      case Types.INTEGER:
                      case Types.TINYINT:
                      case Types.SMALLINT:
                      case Types.BIGINT:
                          // 处理unsigned类型
                          Number number = (Number) value;
                          boolean isUnsigned = (fieldMeta != null ? fieldMeta.isUnsigned() : (existOptionalMetaData ? info.unsigned : false));
                          if (isUnsigned && number.longValue() < 0) {
                              switch (buffer.getLength()) {
                                  case 1: /* MYSQL_TYPE_TINY */
                                      columnBuilder.setValue(String.valueOf(Integer.valueOf(TINYINT_MAX_VALUE
                                                                                            + number.intValue())));
                                      javaType = Types.SMALLINT; // 往上加一个量级
                                      break;
  
                                  case 2: /* MYSQL_TYPE_SHORT */
                                      columnBuilder.setValue(String.valueOf(Integer.valueOf(SMALLINT_MAX_VALUE
                                                                                            + number.intValue())));
                                      javaType = Types.INTEGER; // 往上加一个量级
                                      break;
  
                                  case 3: /* MYSQL_TYPE_INT24 */
                                      columnBuilder.setValue(String.valueOf(Integer.valueOf(MEDIUMINT_MAX_VALUE
                                                                                            + number.intValue())));
                                      javaType = Types.INTEGER; // 往上加一个量级
                                      break;
  
                                  case 4: /* MYSQL_TYPE_LONG */
                                      columnBuilder.setValue(String.valueOf(Long.valueOf(INTEGER_MAX_VALUE
                                                                                         + number.longValue())));
                                      javaType = Types.BIGINT; // 往上加一个量级
                                      break;
  
                                  case 8: /* MYSQL_TYPE_LONGLONG */
                                      columnBuilder.setValue(BIGINT_MAX_VALUE.add(BigInteger.valueOf(number.longValue()))
                                          .toString());
                                      javaType = Types.DECIMAL; // 往上加一个量级，避免执行出错
                                      break;
                              }
                          } else {
                              // 对象为number类型，直接valueof即可
                              columnBuilder.setValue(String.valueOf(value));
                          }
                          break;
                      case Types.REAL: // float
                      case Types.DOUBLE: // double
                          // 对象为number类型，直接valueof即可
                          columnBuilder.setValue(String.valueOf(value));
                          break;
                      case Types.BIT:// bit
                          // 对象为number类型
                          columnBuilder.setValue(String.valueOf(value));
                          break;
                      case Types.DECIMAL:
                          columnBuilder.setValue(((BigDecimal) value).toPlainString());
                          break;
                      case Types.TIMESTAMP:
                          // 修复时间边界值
                          // String v = value.toString();
                          // v = v.substring(0, v.length() - 2);
                          // columnBuilder.setValue(v);
                          // break;
                      case Types.TIME:
                      case Types.DATE:
                          // 需要处理year
                          columnBuilder.setValue(value.toString());
                          break;
                      case Types.BINARY:
                      case Types.VARBINARY:
                      case Types.LONGVARBINARY:
                          // fixed text encoding
                          // https://github.com/AlibabaTech/canal/issues/18
                          // mysql binlog中blob/text都处理为blob类型，需要反查table
                          // meta，按编码解析text
                          if (fieldMeta != null && isText(fieldMeta.getColumnType())) {
                              columnBuilder.setValue(new String((byte[]) value, charset));
                              javaType = Types.CLOB;
                          } else {
                              // byte数组，直接使用iso-8859-1保留对应编码，浪费内存
                              columnBuilder.setValue(new String((byte[]) value, ISO_8859_1));
                              // columnBuilder.setValueBytes(ByteString.copyFrom((byte[])
                              // value));
                              javaType = Types.BLOB;
                          }
                          break;
                      case Types.CHAR:
                      case Types.VARCHAR:
                          columnBuilder.setValue(value.toString());
                          break;
                      default:
                          columnBuilder.setValue(value.toString());
                  }
              }
  
              columnBuilder.setSqlType(javaType);
              // 设置是否update的标记位
              columnBuilder.setUpdated(isAfter
                                       && isUpdate(rowDataBuilder.getBeforeColumnsList(),
                                           columnBuilder.getIsNull() ? null : columnBuilder.getValue(),
                                           i));
              if (needField(fieldList, blackFieldList, columnBuilder.getName())) {
                  if (isAfter) {
                      rowDataBuilder.addAfterColumns(columnBuilder.build());
                  } else {
                      rowDataBuilder.addBeforeColumns(columnBuilder.build());
                  }
              }
          }
  
          return tableError;
  
      }
  
      private Entry buildQueryEntry(String queryString, LogHeader logHeader, String tableName) {
          Header header = createHeader(logHeader, "", tableName, EventType.QUERY);
          RowChange.Builder rowChangeBuider = RowChange.newBuilder();
          rowChangeBuider.setSql(queryString);
          rowChangeBuider.setEventType(EventType.QUERY);
          return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
      }
  
      private Entry buildQueryEntry(String queryString, LogHeader logHeader) {
          Header header = createHeader(logHeader, "", "", EventType.QUERY);
          RowChange.Builder rowChangeBuider = RowChange.newBuilder();
          rowChangeBuider.setSql(queryString);
          rowChangeBuider.setEventType(EventType.QUERY);
          return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString());
      }
  
      private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType) {
          return createHeader(logHeader, schemaName, tableName, eventType, -1);
      }
  
      private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType,
                                  Integer rowsCount) {
          // header会做信息冗余,方便以后做检索或者过滤
          Header.Builder headerBuilder = Header.newBuilder();
          headerBuilder.setVersion(version);
          headerBuilder.setLogfileName(logHeader.getLogFileName());
          // 记录的是该binlog的start offest
          headerBuilder.setLogfileOffset(logHeader.getLogPos() - logHeader.getEventLen());
          headerBuilder.setServerId(logHeader.getServerId());
          headerBuilder.setServerenCode(UTF_8);// 经过java输出后所有的编码为unicode
          headerBuilder.setExecuteTime(logHeader.getWhen() * 1000L);
          headerBuilder.setSourceType(Type.MYSQL);
          if (eventType != null) {
              headerBuilder.setEventType(eventType);
          }
          if (schemaName != null) {
              headerBuilder.setSchemaName(schemaName);
          }
          if (tableName != null) {
              headerBuilder.setTableName(tableName);
          }
          headerBuilder.setEventLength(logHeader.getEventLen());
          // enable gtid position
          if (StringUtils.isNotEmpty(logHeader.getGtidSetStr())) {
              headerBuilder.setGtid(logHeader.getGtidSetStr());
          }
          // add current gtid
          if (StringUtils.isNotEmpty(logHeader.getCurrentGtid())) {
              Pair pair = createSpecialPair("curtGtid", logHeader.getCurrentGtid());
              headerBuilder.addProps(pair);
          }
          // add current gtid sequence no
          if (StringUtils.isNotEmpty(logHeader.getCurrentGtidSn())) {
              Pair pair = createSpecialPair("curtGtidSn", logHeader.getCurrentGtidSn());
              headerBuilder.addProps(pair);
          }
  
          // add current gtid last committed
          if (StringUtils.isNotEmpty(logHeader.getCurrentGtidLastCommit())) {
              Pair pair = createSpecialPair("curtGtidLct", logHeader.getCurrentGtidLastCommit());
              headerBuilder.addProps(pair);
          }
  
          // add rowsCount suppport
          if (rowsCount > 0) {
              Pair pair = createSpecialPair("rowsCount", String.valueOf(rowsCount));
              headerBuilder.addProps(pair);
          }
          return headerBuilder.build();
      }
  
      private boolean isUpdate(List<Column> bfColumns, String newValue, int index) {
          if (bfColumns == null) {
              throw new CanalParseException("ERROR ## the bfColumns is null");
          }
  
          if (index < 0) {
              return false;
          }
  
          for (Column column : bfColumns) {
              if (column.getIndex() == index) {// 比较before / after的column index
                  if (column.getIsNull() && newValue == null) {
                      // 如果全是null
                      return false;
                  } else if (newValue != null && (!column.getIsNull() && column.getValue().equals(newValue))) {
                      // fixed issue #135, old column is Null
                      // 如果不为null，并且相等
                      return false;
                  }
              }
          }
  
          // 比如nolob/minial模式下,可能找不到before记录,认为是有变化
          return true;
      }
  
      private TableMeta getTableMeta(String dbName, String tbName, boolean useCache, EntryPosition position) {
          try {
              return tableMetaCache.getTableMeta(dbName, tbName, useCache, position);
          } catch (Throwable e) {
              String message = ExceptionUtils.getRootCauseMessage(e);
              if (filterTableError) {
                  if (StringUtils.contains(message, "errorNumber=1146") && StringUtils.contains(message, "doesn't exist")) {
                      return null;
                  } else if (StringUtils.contains(message, "errorNumber=1142")
                             && StringUtils.contains(message, "command denied")) {
                      return null;
                  }
              }
  
              throw new CanalParseException(e);
          }
      }
  
      private boolean isText(String columnType) {
          return "LONGTEXT".equalsIgnoreCase(columnType) || "MEDIUMTEXT".equalsIgnoreCase(columnType)
                 || "TEXT".equalsIgnoreCase(columnType) || "TINYTEXT".equalsIgnoreCase(columnType);
      }
  
      private boolean isAliSQLHeartBeat(String schema, String table) {
          return "test".equalsIgnoreCase(schema) && "heartbeat".equalsIgnoreCase(table);
      }
  
      private boolean isRDSHeartBeat(String schema, String table) {
          return "mysql".equalsIgnoreCase(schema) && "ha_health_check".equalsIgnoreCase(table);
      }
  
      /**
       * 字段过滤判断
       */
      private boolean needField(List<String> fieldList, List<String> blackFieldList, String columnName) {
          if (fieldList == null || fieldList.isEmpty()) {
              return blackFieldList == null || blackFieldList.isEmpty()
                     || !blackFieldList.contains(columnName.toUpperCase());
          } else {
              return fieldList.contains(columnName.toUpperCase());
          }
      }
  
      public static TransactionBegin createTransactionBegin(long threadId) {
          TransactionBegin.Builder beginBuilder = TransactionBegin.newBuilder();
          beginBuilder.setThreadId(threadId);
          return beginBuilder.build();
      }
  
      public static TransactionEnd createTransactionEnd(long transactionId) {
          TransactionEnd.Builder endBuilder = TransactionEnd.newBuilder();
          endBuilder.setTransactionId(String.valueOf(transactionId));
          return endBuilder.build();
      }
  
      public static Pair createSpecialPair(String key, String value) {
          Pair.Builder pairBuilder = Pair.newBuilder();
          pairBuilder.setKey(key);
          pairBuilder.setValue(value);
          return pairBuilder.build();
      }
  
      public static Entry createEntry(Header header, EntryType entryType, ByteString storeValue) {
          Entry.Builder entryBuilder = Entry.newBuilder();
          entryBuilder.setHeader(header);
          entryBuilder.setEntryType(entryType);
          entryBuilder.setStoreValue(storeValue);
          return entryBuilder.build();
      }
  
      public void setCharset(Charset charset) {
          this.charset = charset;
      }
  
      public void setNameFilter(AviaterRegexFilter nameFilter) {
          this.nameFilter = nameFilter;
          logger.warn("--> init table filter : " + nameFilter.toString());
      }
  
      public void setNameBlackFilter(AviaterRegexFilter nameBlackFilter) {
          this.nameBlackFilter = nameBlackFilter;
          logger.warn("--> init table black filter : " + nameBlackFilter.toString());
      }
  
      public void setFieldFilterMap(Map<String, List<String>> fieldFilterMap) {
          if (fieldFilterMap != null) {
              this.fieldFilterMap = fieldFilterMap;
          } else {
              this.fieldFilterMap = new HashMap<>();
          }
  
          for (Map.Entry<String, List<String>> entry : this.fieldFilterMap.entrySet()) {
              logger.warn("--> init field filter : " + entry.getKey() + "->" + entry.getValue());
          }
      }
  
      public void setFieldBlackFilterMap(Map<String, List<String>> fieldBlackFilterMap) {
          if (fieldBlackFilterMap != null) {
              this.fieldBlackFilterMap = fieldBlackFilterMap;
          } else {
              this.fieldBlackFilterMap = new HashMap<>();
          }
  
          for (Map.Entry<String, List<String>> entry : this.fieldBlackFilterMap.entrySet()) {
              logger.warn("--> init field black filter : " + entry.getKey() + "->" + entry.getValue());
          }
      }
  
      public void setTableMetaCache(TableMetaCache tableMetaCache) {
          this.tableMetaCache = tableMetaCache;
      }
  
      public void setFilterQueryDcl(boolean filterQueryDcl) {
          this.filterQueryDcl = filterQueryDcl;
      }
  
      public void setFilterQueryDml(boolean filterQueryDml) {
          this.filterQueryDml = filterQueryDml;
      }
  
      public void setFilterQueryDdl(boolean filterQueryDdl) {
          this.filterQueryDdl = filterQueryDdl;
      }
  
      public void setFilterTableError(boolean filterTableError) {
          this.filterTableError = filterTableError;
      }
  
      public void setFilterRows(boolean filterRows) {
          this.filterRows = filterRows;
      }
  
      public void setUseDruidDdlFilter(boolean useDruidDdlFilter) {
          this.useDruidDdlFilter = useDruidDdlFilter;
      }
  }
  