From 0f14ed926bc44f74f38e3e0b54d7dfa2d4c860ca Mon Sep 17 00:00:00 2001 From: wuwo Date: Mon, 29 Oct 2018 17:32:10 +0800 Subject: [PATCH 01/11] pass useDruidDdlFilter parameter to LogEventConvert --- .../canal/parse/inbound/mysql/AbstractMysqlEventParser.java | 1 + .../canal/parse/inbound/mysql/dbsync/LogEventConvert.java | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java index 8050186d..5e94e40b 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java @@ -60,6 +60,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser { convert.setFilterQueryDdl(filterQueryDdl); convert.setFilterRows(filterRows); convert.setFilterTableError(filterTableError); + convert.setUseDruidDdlFilter(useDruidDdlFilter); return convert; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index ec71465b..5a302b86 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -981,4 +981,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar public void setFilterRows(boolean filterRows) { this.filterRows = filterRows; } + + public void setUseDruidDdlFilter(boolean useDruidDdlFilter) { + this.useDruidDdlFilter = useDruidDdlFilter; + } } From 4c7129115c56fca0949b4e046375b5cd498d5bcd Mon Sep 17 00:00:00 2001 From: wuwo Date: Mon, 29 Oct 2018 17:49:34 +0800 Subject: [PATCH 02/11] a little improve about tsdb --- .../inbound/mysql/dbsync/LogEventConvert.java | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 5a302b86..5288ac2a 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -233,7 +233,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar if (useDruidDdlFilter) { List results = DruidDdlParser.parse(queryString, event.getDbName()); for (DdlResult result : results) { - if (!processFilter(queryString, result)) { + if (!processFilter(queryString, result, event, isSeek)) { // 只要有一个数据不进行过滤 notFilter = true; } @@ -246,7 +246,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } else { DdlResult result = SimpleDdlParser.parse(queryString, event.getDbName()); - if (!processFilter(queryString, result)) { + if (!processFilter(queryString, result, event, isSeek)) { notFilter = true; } @@ -260,12 +260,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return null; } - if (!isSeek) { - // 使用新的表结构元数据管理方式 - EntryPosition position = createPosition(event.getHeader()); - tableMetaCache.apply(position, event.getDbName(), queryString, null); - } - Header header = createHeader(event.getHeader(), schemaName, tableName, type); RowChange.Builder rowChangeBuider = RowChange.newBuilder(); if (type != EventType.QUERY && type != EventType.INSERT && type != EventType.UPDATE @@ -285,7 +279,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return StringUtils.substringAfter(queryString, type); } - private boolean processFilter(String queryString, DdlResult result) { + private boolean processFilter(String queryString, DdlResult result, QueryLogEvent event, boolean isSeek) { String schemaName = result.getSchemaName(); String tableName = result.getTableName(); // fixed issue https://github.com/alibaba/canal/issues/58 @@ -312,10 +306,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX || result.getType() == EventType.DINDEX) { // 针对DDL类型 - if (filterQueryDdl) { - return true; - } - if (StringUtils.isEmpty(tableName) || (result.getType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName()))) { // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析 @@ -351,6 +341,16 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } } + + if (!isSeek) { + // 使用新的表结构元数据管理方式 + EntryPosition position = createPosition(event.getHeader()); + tableMetaCache.apply(position, event.getDbName(), queryString, null); + } + + if (filterQueryDdl) { + return true; + } } else if (result.getType() == EventType.INSERT || result.getType() == EventType.UPDATE || result.getType() == EventType.DELETE) { // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤 From 4ce2b94241fcc9fdca6e3c2c97a99a5ecbdb1a8c Mon Sep 17 00:00:00 2001 From: wuwo Date: Tue, 30 Oct 2018 11:49:49 +0800 Subject: [PATCH 03/11] rollback tsdb change --- .../inbound/mysql/dbsync/LogEventConvert.java | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 5288ac2a..5a302b86 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -233,7 +233,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar if (useDruidDdlFilter) { List results = DruidDdlParser.parse(queryString, event.getDbName()); for (DdlResult result : results) { - if (!processFilter(queryString, result, event, isSeek)) { + if (!processFilter(queryString, result)) { // 只要有一个数据不进行过滤 notFilter = true; } @@ -246,7 +246,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } else { DdlResult result = SimpleDdlParser.parse(queryString, event.getDbName()); - if (!processFilter(queryString, result, event, isSeek)) { + if (!processFilter(queryString, result)) { notFilter = true; } @@ -260,6 +260,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return null; } + if (!isSeek) { + // 使用新的表结构元数据管理方式 + EntryPosition position = createPosition(event.getHeader()); + tableMetaCache.apply(position, event.getDbName(), queryString, null); + } + Header header = createHeader(event.getHeader(), schemaName, tableName, type); RowChange.Builder rowChangeBuider = RowChange.newBuilder(); if (type != EventType.QUERY && type != EventType.INSERT && type != EventType.UPDATE @@ -279,7 +285,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return StringUtils.substringAfter(queryString, type); } - private boolean processFilter(String queryString, DdlResult result, QueryLogEvent event, boolean isSeek) { + private boolean processFilter(String queryString, DdlResult result) { String schemaName = result.getSchemaName(); String tableName = result.getTableName(); // fixed issue https://github.com/alibaba/canal/issues/58 @@ -306,6 +312,10 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX || result.getType() == EventType.DINDEX) { // 针对DDL类型 + if (filterQueryDdl) { + return true; + } + if (StringUtils.isEmpty(tableName) || (result.getType() == EventType.RENAME && StringUtils.isEmpty(result.getOriTableName()))) { // 如果解析不出tableName,记录一下日志，方便bugfix，目前直接抛出异常，中断解析 @@ -341,16 +351,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } } } - - if (!isSeek) { - // 使用新的表结构元数据管理方式 - EntryPosition position = createPosition(event.getHeader()); - tableMetaCache.apply(position, event.getDbName(), queryString, null); - } - - if (filterQueryDdl) { - return true; - } } else if (result.getType() == EventType.INSERT || result.getType() == EventType.UPDATE || result.getType() == EventType.DELETE) { // 对外返回，保证兼容，还是返回QUERY类型，这里暂不解析tableName，所以无法支持过滤 From dd8243a56d99f5cee72fe6e7065085c36fb7f767 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Tue, 30 Oct 2018 23:48:05 +0800 Subject: [PATCH 04/11] fixed issue #1075, support message transactor filter --- .../otter/canal/common/CanalMessageSerializer.java | 8 +++++++- .../alibaba/otter/canal/kafka/MessageSerializer.java | 10 +++++++++- .../otter/canal/rocketmq/CanalRocketMQProducer.java | 3 ++- 3 files changed, 18 insertions(+), 3 deletions(-) diff --git a/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java index 665dbf55..fda5dc65 100644 --- a/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java +++ b/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java @@ -16,7 +16,7 @@ import com.google.protobuf.WireFormat; public class CanalMessageSerializer { @SuppressWarnings("deprecation") - public static byte[] serializer(Message data) { + public static byte[] serializer(Message data, boolean filterTransactionEntry) { try { if (data != null) { if (data.getId() != -1) { @@ -53,8 +53,14 @@ public class CanalMessageSerializer { output.checkNoSpaceLeft(); return body; } else if (!CollectionUtils.isEmpty(data.getEntries())) { + // mq模式只会走到非rowEntry模式 CanalPacket.Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder(); for (CanalEntry.Entry entry : data.getEntries()) { + if (filterTransactionEntry + && (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND)) { + continue; + } + messageBuilder.addMessages(entry.toByteString()); } diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java index 77caa0a6..39c297d8 100644 --- a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java +++ b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.kafka; import java.util.Map; +import org.apache.commons.lang.BooleanUtils; import org.apache.kafka.common.serialization.Serializer; import com.alibaba.otter.canal.common.CanalMessageSerializer; @@ -15,13 +16,20 @@ import com.alibaba.otter.canal.protocol.Message; */ public class MessageSerializer implements Serializer { + private boolean filterTransactionEntry = false; + + public MessageSerializer(){ + this.filterTransactionEntry = BooleanUtils.toBoolean(System.getProperty("canal.instance.filter.transaction.entry", + "false")); + } + @Override public void configure(Map configs, boolean isKey) { } @Override public byte[] serialize(String topic, Message data) { - return CanalMessageSerializer.serializer(data); + return CanalMessageSerializer.serializer(data, filterTransactionEntry); } @Override diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java index aaace132..19872490 100644 --- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java +++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java @@ -47,7 +47,8 @@ public class CanalRocketMQProducer implements CanalMQProducer { Callback callback) { if (!mqProperties.getFlatMessage()) { try { - Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data)); + Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data, + mqProperties.isFilterTransactionEntry())); logger.debug("send message:{} to destination:{}, partition: {}", message, destination.getCanalDestination(), From 1d9e2cceddf722728a8614b8204674fc176f41ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 31 Oct 2018 12:51:02 +0800 Subject: [PATCH 05/11] fixed issue #1076 , support rds binlog expired --- .../parse/inbound/mysql/MysqlEventParser.java | 17 +++++++++++- .../mysql/rds/BinlogDownloadQueue.java | 5 ++++ .../mysql/rds/RdsBinlogEventParserProxy.java | 27 +++++++++++++------ .../mysql/rds/RdsLocalBinlogEventParser.java | 10 ++++++- 4 files changed, 49 insertions(+), 10 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index df9c952c..45f2559f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -69,6 +69,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE // update by yishun.chen,特殊异常处理参数 private int dumpErrorCount = 0; // binlogDump失败异常计数 private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值 + private boolean rdsOssMode = false; protected ErosaConnection buildErosaConnection() { return buildMysqlConnection(this.runningInfo); @@ -352,7 +353,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE return logPosition.getPostion(); } - if (masterPosition!=null && StringUtils.isNotEmpty(masterPosition.getGtid())) { + if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) { return masterPosition; } } @@ -493,6 +494,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE dumpErrorCount = 0; return findPosition; } + + Long timestamp = logPosition.getPostion().getTimestamp(); + if (isRdsOssMode() && (timestamp != null && timestamp > 0)) { + // 如果binlog位点不存在，并且属于timestamp不为空,可以返回null走到oss binlog处理 + return null; + } } // 其余情况 logger.warn("prepare to find start position just last position\n {}", @@ -905,4 +912,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE this.dumpErrorCountThreshold = dumpErrorCountThreshold; } + public boolean isRdsOssMode() { + return rdsOssMode; + } + + public void setRdsOssMode(boolean rdsOssMode) { + this.rdsOssMode = rdsOssMode; + } + } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java index e132514b..9a5cd271 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java @@ -41,6 +41,7 @@ import org.apache.http.ssl.TrustStrategy; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile; /** @@ -103,6 +104,9 @@ public class BinlogDownloadQueue { public BinlogFile tryOne() throws Throwable { BinlogFile binlogFile = binlogList.poll(); + if (binlogFile == null) { + throw new CanalParseException("download binlog is null"); + } download(binlogFile); hostId = binlogFile.getHostInstanceID(); this.currentSize++; @@ -162,6 +166,7 @@ public class BinlogDownloadQueue { this.currentSize = 0; binlogList.clear(); downloadQueue.clear(); + downloadThread = null; } private void download(BinlogFile binlogFile) throws Throwable { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java index 3c38ed0c..f3a220b0 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java @@ -22,14 +22,14 @@ import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser; */ public class RdsBinlogEventParserProxy extends MysqlEventParser { - private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址 - private String accesskey; // 云账号的ak - private String secretkey; // 云账号sk - private String instanceId; // rds实例id - private String directory; // binlog目录 - private int batchFileSize = 4; // 最多下载的binlog文件数量 + private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址 + private String accesskey; // 云账号的ak + private String secretkey; // 云账号sk + private String instanceId; // rds实例id + private String directory; // binlog目录 + private int batchFileSize = 4; // 最多下载的binlog文件数量 - private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser(); + private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = null; private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() { @Override @@ -43,8 +43,11 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser { @Override public void start() { - if (StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey) + if (rdsLocalBinlogEventParser == null && StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey) && StringUtils.isNotEmpty(instanceId)) { + rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser(); + // rds oss mode + setRdsOssMode(true); final ParserExceptionHandler targetHandler = this.getParserExceptionHandler(); if (directory == null) { directory = System.getProperty("java.io.tmpdir", "/tmp") + "/" + destination; @@ -119,10 +122,18 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser { long serverId = rdsBinlogEventParserProxy.getServerId(); rdsLocalBinlogEventParser.setServerId(serverId); rdsBinlogEventParserProxy.stop(); + } catch (Throwable e) { + logger.info("handle exception failed", e); + } + + try { logger.info("start rds mysql binlog parser!"); rdsLocalBinlogEventParser.start(); } catch (Throwable e) { logger.info("handle exception failed", e); + rdsLocalBinlogEventParser.stop(); + RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this; + rdsBinlogEventParserProxy.start();// 继续重试 } } }); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java index aac14d99..2e0960fe 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java @@ -57,7 +57,11 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements if (entryPosition == null) { throw new PositionNotFoundException("position not found!"); } - long startTimeInMill = entryPosition.getTimestamp(); + Long startTimeInMill = entryPosition.getTimestamp(); + if (startTimeInMill == null || startTimeInMill <= 0) { + throw new PositionNotFoundException("position timestamp is empty!"); + } + startTime = startTimeInMill; List binlogFiles = RdsBinlogOpenApi.listBinlogFiles(url, accesskey, @@ -65,6 +69,10 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements instanceId, new Date(startTime), new Date(endTime)); + if (binlogFiles.isEmpty()) { + throw new CanalParseException("start timestamp : " + startTimeInMill + " binlog files is empty"); + } + binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchFileSize, directory); binlogDownloadQueue.silenceDownload(); needWait = true; From ff3ccde62a3bc8dac5ae74872f578d9bd0186cb2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Wed, 31 Oct 2018 14:25:54 +0800 Subject: [PATCH 06/11] fixed rds binlog dump --- .../parse/inbound/AbstractEventParser.java | 6 +++++- .../parse/inbound/mysql/MysqlEventParser.java | 19 +++++++++++-------- .../inbound/mysql/dbsync/LogEventConvert.java | 4 ++-- .../mysql/rds/BinlogDownloadQueue.java | 11 +++++++++-- 4 files changed, 27 insertions(+), 13 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index dbb40287..8902c05e 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -353,7 +353,11 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle eventSink.interrupt(); if (multiStageCoprocessor != null && multiStageCoprocessor.isStart()) { - multiStageCoprocessor.stop(); + try { + multiStageCoprocessor.stop(); + } catch (Throwable t) { + logger.debug("multi processor rejected:", t); + } } try { diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 45f2559f..e8ca45e3 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -752,14 +752,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE logPosition.setPostion(entryPosition); } - if (entry == null) { - return true; - } - - String logfilename = entry.getHeader().getLogfileName(); - Long logfileoffset = entry.getHeader().getLogfileOffset(); - Long logposTimestamp = entry.getHeader().getExecuteTime(); - Long serverId = entry.getHeader().getServerId(); + // 直接用event的位点来处理,解决一个binlog文件里没有任何事件导致死循环无法退出的问题 + String logfilename = event.getHeader().getLogFileName(); + // 记录的是binlog end offest, + // 因为与其对比的offest是show master status里的end offest + Long logfileoffset = event.getHeader().getLogPos(); + Long logposTimestamp = event.getHeader().getWhen() * 1000; + Long serverId = event.getHeader().getServerId(); // 如果最小的一条记录都不满足条件，可直接退出 if (logposTimestamp >= startTimestamp) { @@ -771,6 +770,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE return false; } + if (entry == null) { + return true; + } + // 记录一下上一个事务结束的位置，即下一个事务的position // position = current + // data.length，代表该事务的下一条offest，避免多余的事务重复 diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index ec71465b..0b6ab412 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -567,8 +567,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } private EntryPosition createPosition(LogHeader logHeader) { - return new EntryPosition(logHeader.getLogFileName(), - logHeader.getLogPos(), + return new EntryPosition(logHeader.getLogFileName(), logHeader.getLogPos() - logHeader.getEventLen(), // startPos logHeader.getWhen() * 1000L, logHeader.getServerId()); // 记录到秒 } @@ -818,6 +817,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar 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 diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java index 9a5cd271..b90b89d5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java @@ -135,7 +135,7 @@ public class BinlogDownloadQueue { if (StringUtils.isNotEmpty(needCompareName) && StringUtils.endsWith(needCompareName, "tar")) { needCompareName = needCompareName.substring(0, needCompareName.indexOf(".")); } - return fileName.equalsIgnoreCase(needCompareName) && binlogList.isEmpty(); + return (needCompareName == null || fileName.equalsIgnoreCase(needCompareName)) && binlogList.isEmpty(); } public void prepare() throws InterruptedException { @@ -166,7 +166,14 @@ public class BinlogDownloadQueue { this.currentSize = 0; binlogList.clear(); downloadQueue.clear(); - downloadThread = null; + try { + downloadThread.interrupt(); + downloadThread.join();// 等待其结束 + } catch (InterruptedException e) { + // ignore + } finally { + downloadThread = null; + } } private void download(BinlogFile binlogFile) throws Throwable { From 3bb408194ddcf39de74af2c94a43537469dc9752 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Thu, 1 Nov 2018 16:03:40 +0800 Subject: [PATCH 07/11] fixed issue #1081 , support mariadb rotate_event support checksum --- .../parse/inbound/mysql/MysqlConnection.java | 68 +++++++++++++++++-- .../mysql/MysqlMultiStageCoprocessor.java | 16 ++++- .../canal/parse/DirectLogFetcherTest.java | 48 +++++++++++++ 3 files changed, 123 insertions(+), 9 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 4348dc29..e3056cb9 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -37,21 +37,23 @@ import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; public class MysqlConnection implements ErosaConnection { - private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); + private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class); private MysqlConnector connector; private long slaveId; - private Charset charset = Charset.forName("UTF-8"); + private Charset charset = Charset.forName("UTF-8"); private BinlogFormat binlogFormat; private BinlogImage binlogImage; // tsdb releated private AuthenticationInfo authInfo; - protected int connTimeout = 5 * 1000; // 5秒 - protected int soTimeout = 60 * 60 * 1000; // 1小时 + protected int connTimeout = 5 * 1000; // 5秒 + protected int soTimeout = 60 * 60 * 1000; // 1小时 + private int binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; // dump binlog bytes, 暂不包括meta与TSDB private AtomicLong receivedBinlogBytes; @@ -118,7 +120,7 @@ public class MysqlConnection implements ErosaConnection { */ public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); - + loadBinlogChecksum(); sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); @@ -128,6 +130,7 @@ public class MysqlConnection implements ErosaConnection { decoder.handle(LogEvent.QUERY_EVENT); decoder.handle(LogEvent.XID_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); LogEvent event = null; @@ -145,12 +148,14 @@ public class MysqlConnection implements ErosaConnection { public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { updateSettings(); + loadBinlogChecksum(); sendRegisterSlave(); sendBinlogDump(binlogfilename, binlogPosition); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); LogEvent event = null; @@ -173,6 +178,7 @@ public class MysqlConnection implements ErosaConnection { @Override public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException { updateSettings(); + loadBinlogChecksum(); sendBinlogDumpGTID(gtidSet); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); @@ -180,6 +186,7 @@ public class MysqlConnection implements ErosaConnection { fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); // fix bug: #890 将gtid传输至context中，供decode使用 context.setGtidSet(gtidSet); while (fetcher.fetch()) { @@ -207,9 +214,11 @@ public class MysqlConnection implements ErosaConnection { @Override public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException { updateSettings(); + loadBinlogChecksum(); sendRegisterSlave(); sendBinlogDump(binlogfilename, binlogPosition); ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this); + ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum); DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); try { fetcher.start(connector.getChannel()); @@ -234,9 +243,10 @@ public class MysqlConnection implements ErosaConnection { @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()); @@ -482,6 +492,52 @@ public class MysqlConnection implements ErosaConnection { } } + /** + * 获取主库checksum信息 + * + * 
+     * mariadb区别于mysql会在binlog的第一个事件Rotate_Event里也会采用checksum逻辑,而mysql是在第二个binlog事件之后才感知是否需要处理checksum
+     * 导致maraidb只要是开启checksum就会出现binlog文件名解析乱码
+     * fixed issue : https://github.com/alibaba/canal/issues/1081
+     *
+ */ + private void loadBinlogChecksum() { + if (checkMariaDB()) { + ResultSetPacket rs = null; + try { + rs = query("select @@global.binlog_checksum"); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32; + } else { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; + } + } + } + + /** + * 获取是否为mariadb + */ + private boolean checkMariaDB() { + ResultSetPacket rs = null; + try { + rs = query("SELECT @@version"); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1) { + return StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB"); + } + + return false; + } + private void accumulateReceivedBytes(long x) { if (receivedBinlogBytes != null) { receivedBinlogBytes.addAndGet(x); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 3db8ecc4..8b682f5d 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -33,6 +33,7 @@ import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; @@ -69,6 +70,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private WorkerPool workerPool; private BatchEventProcessor simpleParserStage; private BatchEventProcessor sinkStoreStage; + private LogContext logContext; public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert, EventTransactionBuffer transactionBuffer, String destination){ @@ -95,9 +97,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier(); ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler(); // stage 2 + this.logContext = new LogContext(); simpleParserStage = new BatchEventProcessor(disruptorMsgBuffer, sequenceBarrier, - new SimpleParserStage()); + new SimpleParserStage(logContext)); simpleParserStage.setExceptionHandler(exceptionHandler); disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence()); @@ -128,6 +131,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement workerPool.start(parserExecutor); } + public void setBinlogChecksum(int binlogChecksum) { + if (binlogChecksum != LogEvent.BINLOG_CHECKSUM_ALG_OFF) { + logContext.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); + } + } + @Override public void stop() { // fix bug #968，对于pool与 @@ -239,9 +248,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement private LogDecoder decoder; private LogContext context; - public SimpleParserStage(){ + public SimpleParserStage(LogContext context){ decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); - context = new LogContext(); + this.context = context; if (gtidSet != null) { context.setGtidSet(gtidSet); } @@ -468,4 +477,5 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement public void setGtidSet(GTIDSet gtidSet) { this.gtidSet = gtidSet; } + } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java index aa48f9a3..2cb6ed8b 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java @@ -2,6 +2,7 @@ package com.alibaba.otter.canal.parse; import java.io.IOException; import java.net.InetSocketAddress; +import java.util.List; import org.apache.commons.lang.StringUtils; import org.junit.Assert; @@ -10,21 +11,27 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector; +import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor; import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor; import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket; import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket; +import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket; import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager; +import com.alibaba.otter.canal.parse.exception.CanalParseException; import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent; public class DirectLogFetcherTest { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + private boolean isMariaDB; + private int binlogChecksum; @Test public void testSimple() { @@ -33,6 +40,7 @@ public class DirectLogFetcherTest { MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxx", "xxxx"); connector.connect(); updateSettings(connector); + loadBinlogChecksum(connector); sendRegisterSlave(connector, 3); sendBinlogDump(connector, "mysql-bin.000001", 4L, 3); @@ -40,6 +48,8 @@ public class DirectLogFetcherTest { LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); + context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum)); + while (fetcher.fetch()) { LogEvent event = null; event = decoder.decode(fetcher, context); @@ -188,6 +198,44 @@ public class DirectLogFetcherTest { } } + private void loadBinlogChecksum(MysqlConnector connector) { + checkMariaDB(connector); + if (isMariaDB) { + ResultSetPacket rs = null; + try { + rs = query("select @@global.binlog_checksum", connector); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32; + } else { + binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF; + } + } + } + + private void checkMariaDB(MysqlConnector connector) { + ResultSetPacket rs = null; + try { + rs = query("SELECT @@version", connector); + } catch (IOException e) { + throw new CanalParseException(e); + } + + List columnValues = rs.getFieldValues(); + if (columnValues != null && columnValues.size() >= 1) { + isMariaDB = StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB"); + } + } + + public ResultSetPacket query(String cmd, MysqlConnector connector) throws IOException { + MysqlQueryExecutor exector = new MysqlQueryExecutor(connector); + return exector.query(cmd); + } + public void update(String cmd, MysqlConnector connector) throws IOException { MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector); exector.update(cmd); From 332f759f96f7ff7a6c1b771fa94a493c979b6eaa Mon Sep 17 00:00:00 2001 From: wuwo Date: Thu, 1 Nov 2018 17:38:18 +0800 Subject: [PATCH 08/11] filter dml before apply to tsdb --- .../canal/parse/inbound/mysql/dbsync/LogEventConvert.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 74db38fa..0b88af14 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -260,7 +260,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return null; } - if (!isSeek) { + boolean isDml = (type == EventType.INSERT || type == EventType.UPDATE || type == EventType.DELETE); + + if (!isSeek && !isDml) { // 使用新的表结构元数据管理方式 EntryPosition position = createPosition(event.getHeader()); tableMetaCache.apply(position, event.getDbName(), queryString, null); @@ -268,8 +270,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar Header header = createHeader(event.getHeader(), schemaName, tableName, type); RowChange.Builder rowChangeBuider = RowChange.newBuilder(); - if (type != EventType.QUERY && type != EventType.INSERT && type != EventType.UPDATE - && type != EventType.DELETE) { + if (type != EventType.QUERY && !isDml) { rowChangeBuider.setIsDdl(true); } rowChangeBuider.setSql(queryString); From 9473f4e8fdd1c677ee30bead6b8abfd25c0eb14e Mon Sep 17 00:00:00 2001 From: jiacheo Date: Fri, 2 Nov 2018 01:48:17 +0800 Subject: [PATCH 09/11] Update CanalMessageDeserializer.java fix lazyParseEntry 'raw' wrong value. --- .../alibaba/otter/canal/client/CanalMessageDeserializer.java | 2 ++ 1 file changed, 2 insertions(+) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java b/client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java index b2206890..9d24a23b 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/CanalMessageDeserializer.java @@ -32,10 +32,12 @@ public class CanalMessageDeserializer { if (lazyParseEntry) { // byteString result.setRawEntries(messages.getMessagesList()); + result.setRaw(true); } else { for (ByteString byteString : messages.getMessagesList()) { result.addEntry(CanalEntry.Entry.parseFrom(byteString)); } + result.setRaw(false); } return result; } From 8597d9e926c9782c1e5def45947995611cf20b0f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Fri, 2 Nov 2018 12:51:36 +0800 Subject: [PATCH 10/11] fixed more info --- .../parse/driver/mysql/socket/BioSocketChannel.java | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java index 8de2acfd..dfa2783f 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java @@ -88,8 +88,9 @@ public class BioSocketChannel implements SocketChannel { } } if (remain > 0 && accTimeout >= timeout) { - throw new SocketTimeoutException("Timeout occurred, failed to read " + readSize + " bytes in " + timeout - + " milliseconds."); + throw new SocketTimeoutException("Timeout occurred, failed to read total " + readSize + " bytes in " + + timeout + " milliseconds, actual read only " + (readSize - remain) + + " bytes"); } return data; } @@ -120,8 +121,8 @@ public class BioSocketChannel implements SocketChannel { } if (n < len && accTimeout >= timeout) { - throw new SocketTimeoutException("Timeout occurred, failed to read " + len + " bytes in " + timeout - + " milliseconds."); + throw new SocketTimeoutException("Timeout occurred, failed to read total " + len + " bytes in " + timeout + + " milliseconds, actual read only " + n + " bytes"); } } From 6ff57f53f8169caa1202a1d63caca0c699a6ae2b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=83=E9=94=8B?= Date: Mon, 5 Nov 2018 15:06:40 +0800 Subject: [PATCH 11/11] fixed issue #1070, support mysql8.0.13 --- .../tddl/dbsync/binlog/JsonConversion.java | 82 +++-- .../dbsync/binlog/JsonDiffConversion.java | 154 ++++++++ .../taobao/tddl/dbsync/binlog/LogBuffer.java | 4 +- .../taobao/tddl/dbsync/binlog/LogDecoder.java | 8 + .../taobao/tddl/dbsync/binlog/LogEvent.java | 10 +- .../event/FormatDescriptionLogEvent.java | 1 + .../dbsync/binlog/event/QueryLogEvent.java | 101 ++++- .../dbsync/binlog/event/RowsLogBuffer.java | 86 +++-- .../dbsync/binlog/event/RowsLogEvent.java | 25 +- .../dbsync/binlog/event/TableMapLogEvent.java | 348 +++++++++++++++++- .../binlog/event/UpdateRowsLogEvent.java | 7 +- .../dbsync/binlog/DirectLogFetcherTest.java | 1 + .../dbsync/binlog/FileLogFetcherTest.java | 1 + .../parse/driver/mysql/utils/CharsetUtil.java | 258 +++++++++++++ .../parse/driver/mysql/CharsetUtilTest.java | 51 +++ .../mysql/MysqlMultiStageCoprocessor.java | 1 + .../inbound/mysql/dbsync/LogEventConvert.java | 60 ++- .../canal/parse/DirectLogFetcherTest.java | 124 ++++++- .../MysqlBinlogParsePerformanceTest.java | 5 +- 19 files changed, 1215 insertions(+), 112 deletions(-) create mode 100644 dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java create mode 100644 driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java create mode 100644 driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java index 020591f7..e61b6645 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java @@ -1,5 +1,9 @@ package com.taobao.tddl.dbsync.binlog; +import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber2; +import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber4; +import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.usecondsToStr; + /** * 处理下MySQL json二进制转化为可读的字符串 * @@ -330,12 +334,30 @@ public class JsonConversion { long ultime = Math.abs(packed_value); long intpart = ultime >> 24; int frac = (int) (ultime % (1L << 24)); - text = String.format("%s%02d:%02d:%02d", - packed_value >= 0 ? "" : "-", - (int) ((intpart >> 12) % (1 << 10)), - (int) ((intpart >> 6) % (1 << 6)), - (int) (intpart % (1 << 6))); - text = text + "." + usecondsToStr(frac, 6); + // text = String.format("%s%02d:%02d:%02d", + // packed_value >= 0 ? "" : "-", + // (int) ((intpart >> 12) % (1 << 10)), + // (int) ((intpart >> 6) % (1 << 6)), + // (int) (intpart % (1 << 6))); + // text = text + "." + usecondsToStr(frac, 6); + StringBuilder builder = new StringBuilder(17); + if (packed_value < 0) { + builder.append('-'); + } + + int d = (int) ((intpart >> 12) % (1 << 10)); + if (d > 100) { + builder.append(String.valueOf(d)); + } else { + appendNumber2(builder, d); + } + builder.append(':'); + appendNumber2(builder, (int) ((intpart >> 6) % (1 << 6))); + builder.append(':'); + appendNumber2(builder, (int) (intpart % (1 << 6))); + + builder.append('.').append(usecondsToStr(frac, 6)); + text = builder.toString(); } buf.append('"').append(text).append('"'); } else if (m_field_type == LogEvent.MYSQL_TYPE_DATE || m_field_type == LogEvent.MYSQL_TYPE_DATETIME @@ -351,14 +373,28 @@ public class JsonConversion { long ymd = intpart >> 17; long ym = ymd >> 5; long hms = intpart % (1 << 17); - text = String.format("%04d-%02d-%02d %02d:%02d:%02d", - (int) (ym / 13), - (int) (ym % 13), - (int) (ymd % (1 << 5)), - (int) (hms >> 12), - (int) ((hms >> 6) % (1 << 6)), - (int) (hms % (1 << 6))); - text = text + "." + usecondsToStr(frac, 6); + // text = + // String.format("%04d-%02d-%02d %02d:%02d:%02d", + // (int) (ym / 13), + // (int) (ym % 13), + // (int) (ymd % (1 << 5)), + // (int) (hms >> 12), + // (int) ((hms >> 6) % (1 << 6)), + // (int) (hms % (1 << 6))); + StringBuilder builder = new StringBuilder(26); + appendNumber4(builder, (int) (ym / 13)); + builder.append('-'); + appendNumber2(builder, (int) (ym % 13)); + builder.append('-'); + appendNumber2(builder, (int) (ymd % (1 << 5))); + builder.append(' '); + appendNumber2(builder, (int) (hms >> 12)); + builder.append(':'); + appendNumber2(builder, (int) ((hms >> 6) % (1 << 6))); + builder.append(':'); + appendNumber2(builder, (int) (hms % (1 << 6))); + builder.append('.').append(usecondsToStr(frac, 6)); + text = builder.toString(); } buf.append('"').append(text).append('"'); } else { @@ -397,22 +433,4 @@ public class JsonConversion { OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR } - private static String usecondsToStr(int frac, int meta) { - String sec = String.valueOf(frac); - if (meta > 6) { - throw new IllegalArgumentException("unknow useconds meta : " + meta); - } - - if (sec.length() < 6) { - StringBuilder result = new StringBuilder(6); - int len = 6 - sec.length(); - for (; len > 0; len--) { - result.append('0'); - } - result.append(sec); - sec = result.toString(); - } - - return sec.substring(0, meta); - } } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java new file mode 100644 index 00000000..223b5d9c --- /dev/null +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java @@ -0,0 +1,154 @@ +package com.taobao.tddl.dbsync.binlog; + +import java.util.ArrayList; +import java.util.List; + +import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value; +import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_enum_type; + +/** + * 处理mysql8.0 parital json diff解析 + * + * @author agapple 2018年11月4日 下午3:53:46 + * @since 1.1.2 + */ +public class JsonDiffConversion { + + /** + * The JSON value in the given path is replaced with a new value. It has the + * same effect as `JSON_REPLACE(col, path, value)`. + */ + public static final int DIFF_OPERATION_REPLACE = 0; + /** + * Add a new element at the given path. If the path specifies an array + * element, it has the same effect as `JSON_ARRAY_INSERT(col, path, value)`. + * If the path specifies an object member, it has the same effect as + * `JSON_INSERT(col, path, value)`. + */ + public static final int DIFF_OPERATION_INSERT = 1; + /** + * The JSON value at the given path is removed from an array or object. It + * has the same effect as `JSON_REMOVE(col, path)`. + */ + public static final int DIFF_OPERATION_REMOVE = 2; + + public static final int JSON_DIFF_OPERATION_COUNT = 3; + + public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex, + String charsetName) { + int position = buffer.position(); + List operation_names = new ArrayList(); + while (buffer.hasRemaining()) { + int operation_int = buffer.getUint8(); + if (operation_int >= JSON_DIFF_OPERATION_COUNT) { + throw new IllegalArgumentException("reading operation type (invalid operation code)"); + } + + // skip path + long path_length = buffer.getPackedLong(); + if (path_length > len) { + throw new IllegalArgumentException("skipping path"); + } + + // compute operation name + byte[] lastP = buffer.getData(buffer.position() + (int) path_length - 1, 1); + String operation_name = json_diff_operation_name(operation_int, lastP[0]); + operation_names.add(operation_name); + + buffer.forward((int) path_length); + // skip value + if (operation_int != DIFF_OPERATION_REMOVE) { + long value_length = buffer.getPackedLong(); + if (value_length > len) { + throw new IllegalArgumentException("skipping path"); + } + + buffer.forward((int) value_length); + } + } + + // Print function names in reverse order. + StringBuilder builder = new StringBuilder(); + for (int i = operation_names.size() - 1; i >= 0; i--) { + if (i == 0 || operation_names.get(i - 1) != operation_names.get(i)) { + builder.append(operation_names.get(i)).append("("); + } + } + + // Print column id + if (columnName != null) { + builder.append(columnName); + } else { + builder.append("@").append(columnIndex); + } + + // In case this vector is empty (a no-op), make an early return + // after printing only the column name + if (operation_names.size() == 0) { + return builder; + } + + // Print comma between column name and next function argument + builder.append(", "); + // Print paths and values. + buffer.position(position); + int diff_i = 0; + while (buffer.hasRemaining()) { + // Read operation + int operation_int = buffer.getUint8(); + + // Read path length + long path_length = buffer.getPackedLong(); + // Print path + builder.append('\'').append(buffer.getFixString((int) path_length)).append('\''); + + if (operation_int != DIFF_OPERATION_REMOVE) { + // Print comma between path and value + builder.append(", "); + // Read value length + long value_length = buffer.getPackedLong(); + + Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), + buffer, + value_length - 1, + charsetName); + buffer.forward((int) value_length - 1); + // Read value + if (jsonValue.m_type == Json_enum_type.ERROR) { + throw new IllegalArgumentException("parsing json value"); + } + StringBuilder jsonBuilder = new StringBuilder(); + jsonValue.toJsonString(jsonBuilder, charsetName); + builder.append(jsonBuilder); + } + + // Print closing parenthesis + if (!buffer.hasRemaining() || operation_names.get(diff_i + 1) != operation_names.get(diff_i)) { + builder.append(")"); + } + + if (buffer.hasRemaining()) { + builder.append(", "); + } + diff_i++; + } + + return builder; + } + + private static String json_diff_operation_name(int operationType, int last_path_char) { + switch (operationType) { + case DIFF_OPERATION_REPLACE: + return "JSON_REPLACE"; + case DIFF_OPERATION_INSERT: + if (last_path_char == ']') { + return "JSON_ARRAY_INSERT"; + } else { + return "JSON_INSERT"; + } + case DIFF_OPERATION_REMOVE: + return "JSON_REMOVE"; + } + throw new IllegalArgumentException("illeagal operationType : " + operationType); + } +} diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java index be61c99f..408eaa4e 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java @@ -1484,7 +1484,9 @@ public class LogBuffer { for (int bit = 0; bit < len; bit += 8) { int flag = ((int) buf[pos++]) & 0xff; - if (flag == 0) continue; + if (flag == 0) { + continue; + } if ((flag & 0x01) != 0) bitmap.set(bit); if ((flag & 0x02) != 0) bitmap.set(bit + 1); if ((flag & 0x04) != 0) bitmap.set(bit + 2); diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java index 639e9875..41739cf8 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java @@ -372,6 +372,14 @@ public final class LogDecoder { header.putGtid(context.getGtidSet(), gtidLogEvent); return event; } + case LogEvent.PARTIAL_UPDATE_ROWS_EVENT: { + RowsLogEvent event = new UpdateRowsLogEvent(header, buffer, descriptionEvent, true); + /* updating position in context */ + logPosition.position = header.getLogPos(); + event.fillTable(context); + header.putGtid(context.getGtidSet(), gtidLogEvent); + return event; + } case LogEvent.GTID_LOG_EVENT: case LogEvent.ANONYMOUS_GTID_LOG_EVENT: { GtidLogEvent event = new GtidLogEvent(header, buffer, descriptionEvent); diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java index 8a5957d0..41208516 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java @@ -169,9 +169,15 @@ public abstract class LogEvent { /* Prepared XA transaction terminal event similar to Xid */ public static final int XA_PREPARE_LOG_EVENT = 38; + /** + * Extension of UPDATE_ROWS_EVENT, allowing partial values according to + * binlog_row_value_options. + */ + public static final int PARTIAL_UPDATE_ROWS_EVENT = 39; + // mariaDb 5.5.34 /* New MySQL/Sun events are to be added right above this comment */ - public static final int MYSQL_EVENTS_END = 39; + public static final int MYSQL_EVENTS_END = 49; public static final int MARIA_EVENTS_BEGIN = 160; /* New Maria event numbers start from here */ @@ -361,6 +367,8 @@ public abstract class LogEvent { return "Anonymous_Gtid"; case PREVIOUS_GTIDS_LOG_EVENT: return "Previous_gtids"; + case PARTIAL_UPDATE_ROWS_EVENT: + return "Update_rows_partial"; default: return "Unknown"; /* impossible */ } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java index 8492e2d7..a14b9111 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/FormatDescriptionLogEvent.java @@ -211,6 +211,7 @@ public final class FormatDescriptionLogEvent extends StartLogEventV3 { postHeaderLen[TRANSACTION_CONTEXT_EVENT - 1] = TRANSACTION_CONTEXT_HEADER_LEN; postHeaderLen[VIEW_CHANGE_EVENT - 1] = VIEW_CHANGE_HEADER_LEN; postHeaderLen[XA_PREPARE_LOG_EVENT - 1] = XA_PREPARE_HEADER_LEN; + postHeaderLen[PARTIAL_UPDATE_ROWS_EVENT - 1] = ROWS_HEADER_LEN_V2; // mariadb 10 postHeaderLen[ANNOTATE_ROWS_EVENT - 1] = ANNOTATE_ROWS_HEADER_LEN; diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java index f6e0459c..c732da99 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java @@ -1,6 +1,7 @@ package com.taobao.tddl.dbsync.binlog.event; import java.io.IOException; +import java.math.BigInteger; import java.nio.charset.Charset; import com.taobao.tddl.dbsync.binlog.CharsetConversion; @@ -285,12 +286,12 @@ public class QueryLogEvent extends LogEvent { * type, * sql_mode */ - + 1 + 1 + 255 /* - * type, - * length - * , - * catalog - */ + + 1 + 1 + 255/* + * type, + * length + * , + * catalog + */ + 1 + 4 /* * type, * auto_increment @@ -330,14 +331,32 @@ public class QueryLogEvent extends LogEvent { * MariaDb type, * sec_part of NOW() */ - + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 + 1 + 32 * 3 - + 1 + 60/* - * type , - * user_len , - * user , - * host_len , - * host - */); + + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 /* + * type + * , + * microseconds + */+ 1 + 32 + * 3 + 1 + 60/* + * type , + * user_len + * , user , + * host_len + * , host + */) + + 1 + 1 /* + * type, + * explicit_def + * ..ts + */+ 1 + 8 /* + * type, + * xid + * of + * DDL + */+ 1 + 2 /* + * type + * , + * default_collation_for_utf8mb4_number + */+ 1 /* sql_require_primary_key */; /** * Fixed data part: *
    @@ -394,7 +413,7 @@ public class QueryLogEvent extends LogEvent { // inspection by the DBA private final long execTime; private final int errorCode; - private final long sessionId; /* thread_id */ + private final long sessionId; /* thread_id */ /** * 'flags2' is a second set of flags (on top of those in Log_event), for @@ -412,6 +431,8 @@ public class QueryLogEvent extends LogEvent { private int clientCharset = -1; private int clientCollation = -1; private int serverCollation = -1; + private int tvSec = -1; + private BigInteger ddlXid = BigInteger.valueOf(-1L); private String charsetName; private String timezone; @@ -562,6 +583,21 @@ public class QueryLogEvent extends LogEvent { */ public static final int Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP = 16; + /** + * The variable carries xid info of 2pc-aware (recoverable) DDL queries. + */ + public static final int Q_DDL_LOGGED_WITH_XID = 17; + /** + * This variable stores the default collation for the utf8mb4 character set. + * Used to support cross-version replication. + */ + public static final int Q_DEFAULT_COLLATION_FOR_UTF8MB4 = 18; + + /** + * Replicate sql_require_primary_key. + */ + public static final int Q_SQL_REQUIRE_PRIMARY_KEY = 19; + /** * FROM MariaDB 5.5.34 */ @@ -625,7 +661,7 @@ public class QueryLogEvent extends LogEvent { break; case Q_MICROSECONDS: // when.tv_usec= uint3korr(pos); - buffer.forward(3); + tvSec = buffer.getInt24(); break; case Q_UPDATED_DB_NAMES: int mtsAccessedDbs = buffer.getUint8(); @@ -646,6 +682,19 @@ public class QueryLogEvent extends LogEvent { } break; case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP: + // thd->variables.explicit_defaults_for_timestamp + buffer.forward(1); + break; + case Q_DDL_LOGGED_WITH_XID: + ddlXid = buffer.getUlong64(); + break; + case Q_DEFAULT_COLLATION_FOR_UTF8MB4: + // int2store(start, + // default_collation_for_utf8mb4_number); + buffer.forward(2); + break; + case Q_SQL_REQUIRE_PRIMARY_KEY: + // *start++ = thd->variables.sql_require_primary_key; buffer.forward(1); break; case Q_HRNOW: @@ -657,8 +706,10 @@ public class QueryLogEvent extends LogEvent { * That's why you must write status vars in growing * order of code */ - if (logger.isDebugEnabled()) logger.debug("Query_log_event has unknown status vars (first has code: " - + code + "), skipping the rest of them"); + if (logger.isDebugEnabled()) { + logger.debug("Query_log_event has unknown status vars (first has code: " + code + + "), skipping the rest of them"); + } break; // Break loop } } @@ -695,6 +746,12 @@ public class QueryLogEvent extends LogEvent { return "Q_UPDATED_DB_NAMES"; case Q_MICROSECONDS: return "Q_MICROSECONDS"; + case Q_DDL_LOGGED_WITH_XID: + return "Q_DDL_LOGGED_WITH_XID"; + case Q_DEFAULT_COLLATION_FOR_UTF8MB4: + return "Q_DEFAULT_COLLATION_FOR_UTF8MB4"; + case Q_SQL_REQUIRE_PRIMARY_KEY: + return "Q_SQL_REQUIRE_PRIMARY_KEY"; } return "CODE#" + code; } @@ -777,6 +834,14 @@ public class QueryLogEvent extends LogEvent { return serverCollation; } + public int getTvSec() { + return tvSec; + } + + public BigInteger getDdlXid() { + return ddlXid; + } + /** * Returns the sql_mode value. *

    diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java index 8cd57332..013494d5 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java @@ -10,6 +10,7 @@ import org.apache.commons.logging.LogFactory; import com.taobao.tddl.dbsync.binlog.JsonConversion; import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value; +import com.taobao.tddl.dbsync.binlog.JsonDiffConversion; import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogEvent; @@ -31,22 +32,33 @@ public final class RowsLogBuffer { private final LogBuffer buffer; private final int columnLen; + private final int jsonColumnCount; private final String charsetName; - // private Calendar cal; private final BitSet nullBits; private int nullBitIndex; + // Read value_options if this is AI for PARTIAL_UPDATE_ROWS_EVENT + private final boolean partial; + private final BitSet partialBits; + private boolean fNull; private int javaType; private int length; private Serializable value; - public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName){ + 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); + } + + public final boolean nextOneRow(BitSet columns) { + return nextOneRow(columns, false); } /** @@ -55,18 +67,30 @@ public final class RowsLogBuffer { * @see mysql-5.1.60/sql/log_event.cc - * Rows_log_event::print_verbose_one_row */ - public final boolean nextOneRow(BitSet columns) { + public final boolean nextOneRow(BitSet columns, boolean after) { final boolean hasOneRow = buffer.hasRemaining(); if (hasOneRow) { int column = 0; for (int i = 0; i < columnLen; i++) - if (columns.get(i)) column++; + if (columns.get(i)) { + column++; + } + if (after && partial) { + partialBits.clear(); + long valueOptions = buffer.getPackedLong(); + int PARTIAL_JSON_UPDATES = 1; + if ((valueOptions & PARTIAL_JSON_UPDATES) != 0) { + partialBits.set(1); + buffer.forward((jsonColumnCount + 7) / 8); + } + } nullBitIndex = 0; nullBits.clear(); buffer.fillBitmap(nullBits, column); + } return hasOneRow; } @@ -77,8 +101,8 @@ public final class RowsLogBuffer { * @see mysql-5.1.60/sql/log_event.cc - * Rows_log_event::print_verbose_one_row */ - public final Serializable nextValue(final int type, final int meta) { - return nextValue(type, meta, false); + public final Serializable nextValue(final String columName, final int columnIndex, final int type, final int meta) { + return nextValue(columName, columnIndex, type, meta, false); } /** @@ -87,7 +111,8 @@ public final class RowsLogBuffer { * @see mysql-5.1.60/sql/log_event.cc - * Rows_log_event::print_verbose_one_row */ - public final Serializable nextValue(final int type, final int meta, boolean isBinary) { + public final Serializable nextValue(final String columName, final int columnIndex, final int type, final int meta, + boolean isBinary) { fNull = nullBits.get(nullBitIndex++); if (fNull) { @@ -97,7 +122,7 @@ public final class RowsLogBuffer { return null; } else { // Extracting field value from packed buffer. - return fetchValue(type, meta, isBinary); + return fetchValue(columName, columnIndex, type, meta, isBinary); } } @@ -248,7 +273,7 @@ public final class RowsLogBuffer { * * @see mysql-5.1.60/sql/log_event.cc - log_event_print_value */ - final Serializable fetchValue(int type, final int meta, boolean isBinary) { + final Serializable fetchValue(String columnName, int columnIndex, int type, final int meta, boolean isBinary) { int len = 0; if (type == LogEvent.MYSQL_TYPE_STRING) { @@ -610,7 +635,7 @@ public final class RowsLogBuffer { // (u32 % 10000) / 100, // u32 % 100); - StringBuilder builder = new StringBuilder(12); + StringBuilder builder = new StringBuilder(17); if (i32 < 0) { builder.append('-'); } @@ -1041,17 +1066,34 @@ public final class RowsLogBuffer { default: throw new IllegalArgumentException("!! Unknown JSON packlen = " + meta); } - if (0 == len) { - // fixed issue #1 by lava, json column of zero length has no - // value, value parsing should be skipped - value = ""; - } else { + + if (partialBits.get(1)) { + // print_json_diff int position = buffer.position(); - Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charsetName); - StringBuilder builder = new StringBuilder(); - jsonValue.toJsonString(builder, charsetName); + StringBuilder builder = JsonDiffConversion.print_json_diff(buffer, + len, + columnName, + columnIndex, + charsetName); value = builder.toString(); buffer.position(position + len); + } else { + if (0 == len) { + // fixed issue #1 by lava, json column of zero length + // has no + // value, value parsing should be skipped + value = ""; + } else { + int position = buffer.position(); + Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), + buffer, + len - 1, + charsetName); + StringBuilder builder = new StringBuilder(); + jsonValue.toJsonString(builder, charsetName); + value = builder.toString(); + buffer.position(position + len); + } } javaType = Types.VARCHAR; length = len; @@ -1120,7 +1162,7 @@ public final class RowsLogBuffer { return length; } - private String usecondsToStr(int frac, int meta) { + public static String usecondsToStr(int frac, int meta) { String sec = String.valueOf(frac); if (meta > 6) { throw new IllegalArgumentException("unknow useconds meta : " + meta); @@ -1139,7 +1181,7 @@ public final class RowsLogBuffer { return sec.substring(0, meta); } - private void appendNumber4(StringBuilder builder, int d) { + public static void appendNumber4(StringBuilder builder, int d) { if (d >= 1000) { builder.append(digits[d / 1000]) .append(digits[(d / 100) % 10]) @@ -1151,7 +1193,7 @@ public final class RowsLogBuffer { } } - private void appendNumber3(StringBuilder builder, int d) { + public static void appendNumber3(StringBuilder builder, int d) { if (d >= 100) { builder.append(digits[d / 100]).append(digits[(d / 10) % 10]).append(digits[d % 10]); } else { @@ -1160,7 +1202,7 @@ public final class RowsLogBuffer { } } - private void appendNumber2(StringBuilder builder, int d) { + public static void appendNumber2(StringBuilder builder, int d) { if (d >= 10) { builder.append(digits[(d / 10) % 10]).append(digits[d % 10]); } else { diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java index f2ce79b0..f11b0734 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java @@ -5,6 +5,7 @@ import java.util.BitSet; import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo; /** * Common base class for all row-containing log events. @@ -61,6 +62,7 @@ public abstract class RowsLogEvent extends LogEvent { /** Bitmap denoting columns available */ protected final int columnLen; + protected final boolean partial; protected final BitSet columns; /** @@ -71,6 +73,8 @@ public abstract class RowsLogEvent extends LogEvent { */ protected final BitSet changeColumns; + protected int jsonColumnCount = 0; + /** XXX: Don't handle buffer in another thread. */ private final LogBuffer rowsBuf; /* * The rows in @@ -109,6 +113,10 @@ public abstract class RowsLogEvent extends LogEvent { public static final int RW_V_EXTRAINFO_TAG = 0; public RowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ + this(header, buffer, descriptionEvent, false); + } + + public RowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent, boolean partial){ super(header); final int commonHeaderLen = descriptionEvent.commonHeaderLen; @@ -153,9 +161,11 @@ public abstract class RowsLogEvent extends LogEvent { buffer.position(commonHeaderLen + postHeaderLen + headerLen); columnLen = (int) buffer.getPackedLong(); + this.partial = partial; columns = buffer.getBitmap(columnLen); - if (header.type == UPDATE_ROWS_EVENT_V1 || header.type == UPDATE_ROWS_EVENT) { + if (header.type == UPDATE_ROWS_EVENT_V1 || header.type == UPDATE_ROWS_EVENT + || header.type == PARTIAL_UPDATE_ROWS_EVENT) { changeColumns = buffer.getBitmap(columnLen); } else { changeColumns = columns; @@ -175,6 +185,17 @@ public abstract class RowsLogEvent extends LogEvent { // delete original table map events stored in the map). context.clearAllTables(); } + + int jsonColumnCount = 0; + int columnCnt = table.getColumnCnt(); + ColumnInfo[] columnInfo = table.getColumnInfo(); + for (int i = 0; i < columnCnt; i++) { + ColumnInfo info = columnInfo[i]; + if (info.type == LogEvent.MYSQL_TYPE_JSON) { + jsonColumnCount++; + } + } + this.jsonColumnCount = jsonColumnCount; } public final long getTableId() { @@ -194,7 +215,7 @@ public abstract class RowsLogEvent extends LogEvent { } public final RowsLogBuffer getRowsBuf(String charsetName) { - return new RowsLogBuffer(rowsBuf, columnLen, charsetName); + return new RowsLogBuffer(rowsBuf, columnLen, charsetName, jsonColumnCount, partial); } public final int getFlags(final int flags) { diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java index e6550b27..18201ea9 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java @@ -1,6 +1,8 @@ package com.taobao.tddl.dbsync.binlog.event; +import java.util.ArrayList; import java.util.BitSet; +import java.util.List; import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogEvent; @@ -88,6 +90,17 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; * first byte, the ninth is in the least significant bit of the second byte, and * so on. * + * + * optional metadata fields + * optional metadata fields are stored in Type, Length, Value(TLV) format. + * Type takes 1 byte. Length is a packed integer value. Values takes Length + * bytes. + * There are some optional metadata defined. They are listed in the table + * + * @ref Table_table_map_event_optional_metadata. Optional metadata fields follow + * null_bits. Whether binlogging an optional metadata is decided by the server. + * The order is not defined, so they can be binlogged in any order. + * * * The table below lists all column types, along with the numerical identifier * for it and the size and interpretation of meta-data used to describe the @@ -284,7 +297,6 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; * of the geometry: 1, 2, 3, or 4. * * - * * @author Changyuan.lh * @version 1.0 */ @@ -322,20 +334,63 @@ public final class TableMapLogEvent extends LogEvent { */ public static final class ColumnInfo { - public int type; - public int meta; + public int type; + public int meta; + public String name; + public boolean unsigned; + public boolean pk; + public List set_enum_values; + public int charset; // 可以通过CharsetUtil进行转化 + public int geoType; + public boolean nullable; + + @Override + public String toString() { + return "ColumnInfo [type=" + type + ", meta=" + meta + ", name=" + name + ", unsigned=" + unsigned + + ", pk=" + pk + ", set_enum_values=" + set_enum_values + ", charset=" + charset + ", geoType=" + + geoType + ", nullable=" + nullable + "]"; + } } protected final int columnCnt; - protected final ColumnInfo[] columnInfo; // buffer for field - // metadata + protected final ColumnInfo[] columnInfo; // buffer + // for + // field + // metadata protected final long tableId; protected BitSet nullBits; /** TM = "Table Map" */ - public static final int TM_MAPID_OFFSET = 0; - public static final int TM_FLAGS_OFFSET = 6; + public static final int TM_MAPID_OFFSET = 0; + public static final int TM_FLAGS_OFFSET = 6; + + // UNSIGNED flag of numeric columns + public static final int SIGNEDNESS = 1; + // Default character set of string columns + public static final int DEFAULT_CHARSET = 2; + // Character set of string columns + public static final int COLUMN_CHARSET = 3; + public static final int COLUMN_NAME = 4; + // String value of SET columns + public static final int SET_STR_VALUE = 5; + // String value of ENUM columns + public static final int ENUM_STR_VALUE = 6; + // Real type of geometry columns + public static final int GEOMETRY_TYPE = 7; + // Primary key without prefix + public static final int SIMPLE_PRIMARY_KEY = 8; + // Primary key with prefix + public static final int PRIMARY_KEY_WITH_PREFIX = 9; + + private int default_charset; + private boolean existOptionalMetaData = false; + + private static final class Pair { + + public int col_index; + public int col_charset; + } /** * Constructor used by slave to read the event from the binary log. @@ -379,7 +434,91 @@ public final class TableMapLogEvent extends LogEvent { final int fieldSize = (int) buffer.getPackedLong(); decodeFields(buffer, fieldSize); nullBits = buffer.getBitmap(columnCnt); + + for (int i = 0; i < columnCnt; i++) { + if (nullBits.get(i)) { + columnInfo[i].nullable = true; + } + } + /* + * After null_bits field, there are some new fields for extra + * metadata. + */ + existOptionalMetaData = false; + List defaultCharsetPairs = null; + List columnCharsets = null; + while (buffer.hasRemaining()) { + // optional metadata fields + int type = buffer.getUint8(); + int len = (int) buffer.getPackedLong(); + + switch (type) { + case SIGNEDNESS: + parse_signedness(buffer, len); + break; + case DEFAULT_CHARSET: + defaultCharsetPairs = parse_default_charset(buffer, len); + break; + case COLUMN_CHARSET: + columnCharsets = parse_column_charset(buffer, len); + break; + case COLUMN_NAME: + // set @@global.binlog_row_metadata='FULL' + // 主要是补充列名相关信息 + existOptionalMetaData = true; + parse_column_name(buffer, len); + break; + case SET_STR_VALUE: + parse_set_str_value(buffer, len, true); + break; + case ENUM_STR_VALUE: + parse_set_str_value(buffer, len, false); + break; + case GEOMETRY_TYPE: + parse_geometry_type(buffer, len); + break; + case SIMPLE_PRIMARY_KEY: + parse_simple_pk(buffer, len); + break; + case PRIMARY_KEY_WITH_PREFIX: + parse_pk_with_prefix(buffer, len); + break; + default: + throw new IllegalArgumentException("unknow type : " + type); + } + } + + if (existOptionalMetaData) { + int index = 0; + int char_col_index = 0; + for (int i = 0; i < columnCnt; i++) { + int cs = -1; + int type = getRealType(columnInfo[i].type, columnInfo[i].meta); + if (is_character_type(type)) { + if (defaultCharsetPairs != null && !defaultCharsetPairs.isEmpty()) { + if (index < defaultCharsetPairs.size() + && char_col_index == defaultCharsetPairs.get(index).col_index) { + cs = defaultCharsetPairs.get(index).col_charset; + index++; + } else { + cs = default_charset; + } + + char_col_index++; + } else if (columnCharsets != null) { + cs = columnCharsets.get(index); + index++; + } + + columnInfo[i].charset = cs; + } + } + } } + + // for (int i = 0; i < columnCnt; i++) { + // System.out.println(columnInfo[i]); + // } } /** @@ -459,6 +598,192 @@ public final class TableMapLogEvent extends LogEvent { buffer.limit(limit); } + private void parse_signedness(LogBuffer buffer, int length) { + // stores the signedness flags extracted from field + List datas = new ArrayList(); + 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 List parse_default_charset(LogBuffer buffer, int length) { + // stores collation numbers extracted from field. + int limit = buffer.position() + length; + this.default_charset = (int) buffer.getPackedLong(); + List datas = new ArrayList(); + while (buffer.hasRemaining() && buffer.position() < limit) { + int col_index = (int) buffer.getPackedLong(); + int col_charset = (int) buffer.getPackedLong(); + + Pair pair = new Pair(); + pair.col_index = col_index; + pair.col_charset = col_charset; + datas.add(pair); + } + + return datas; + } + + private List parse_column_charset(LogBuffer buffer, int length) { + // stores collation numbers extracted from field. + int limit = buffer.position() + length; + List datas = new ArrayList(); + 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); + } + } + + 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> datas = new ArrayList>(); + while (buffer.hasRemaining() && buffer.position() < limit) { + int count = (int) buffer.getPackedLong(); + List data = new ArrayList(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++; + } + } + } + + private void parse_geometry_type(LogBuffer buffer, int length) { + // stores geometry column's types extracted from field. + int limit = buffer.position() + length; + + List datas = new ArrayList(); + 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++; + } + } + } + + 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; + } + } + + 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; + } + } + + private boolean is_numeric_type(int type) { + switch (type) { + case MYSQL_TYPE_TINY: + case MYSQL_TYPE_SHORT: + case MYSQL_TYPE_INT24: + case MYSQL_TYPE_LONG: + case MYSQL_TYPE_LONGLONG: + case MYSQL_TYPE_NEWDECIMAL: + case MYSQL_TYPE_FLOAT: + case MYSQL_TYPE_DOUBLE: + return true; + default: + return false; + } + } + + private boolean is_character_type(int type) { + switch (type) { + case MYSQL_TYPE_STRING: + case MYSQL_TYPE_VAR_STRING: + case MYSQL_TYPE_VARCHAR: + case MYSQL_TYPE_BLOB: + return true; + default: + return false; + } + } + + private int getRealType(int type, int meta) { + if (type == LogEvent.MYSQL_TYPE_STRING) { + if (meta >= 256) { + int byte0 = meta >> 8; + if ((byte0 & 0x30) != 0x30) { + /* a long CHAR() field: see #37426 */ + type = byte0 | 0x30; + } else { + switch (byte0) { + case LogEvent.MYSQL_TYPE_SET: + case LogEvent.MYSQL_TYPE_ENUM: + case LogEvent.MYSQL_TYPE_STRING: + type = byte0; + } + } + } + } + + return type; + } + public final String getDbName() { return dbname; } @@ -478,4 +803,13 @@ public final class TableMapLogEvent extends LogEvent { public final long getTableId() { return tableId; } + + public boolean isExistOptionalMetaData() { + return existOptionalMetaData; + } + + public void setExistOptionalMetaData(boolean existOptional) { + this.existOptionalMetaData = existOptional; + } + } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java index f5e5866e..b604e04d 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java @@ -14,6 +14,11 @@ import com.taobao.tddl.dbsync.binlog.LogBuffer; public final class UpdateRowsLogEvent extends RowsLogEvent { public UpdateRowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){ - super(header, buffer, descriptionEvent); + super(header, buffer, descriptionEvent, false); + } + + public UpdateRowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent, + boolean partial){ + super(header, buffer, descriptionEvent, partial); } } diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java index c48082df..3648da68 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java @@ -45,6 +45,7 @@ public class DirectLogFetcherTest extends BaseLogFetcherTest { parseRowsEvent((WriteRowsLogEvent) event); break; case LogEvent.UPDATE_ROWS_EVENT_V1: + case LogEvent.PARTIAL_UPDATE_ROWS_EVENT: case LogEvent.UPDATE_ROWS_EVENT: parseRowsEvent((UpdateRowsLogEvent) event); break; diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java index 7673f71b..aada3de8 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java @@ -53,6 +53,7 @@ public class FileLogFetcherTest extends BaseLogFetcherTest { parseRowsEvent((WriteRowsLogEvent) event); break; case LogEvent.UPDATE_ROWS_EVENT_V1: + case LogEvent.PARTIAL_UPDATE_ROWS_EVENT: case LogEvent.UPDATE_ROWS_EVENT: parseRowsEvent((UpdateRowsLogEvent) event); break; diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java new file mode 100644 index 00000000..ba98a87f --- /dev/null +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java @@ -0,0 +1,258 @@ +package com.alibaba.otter.canal.parse.driver.mysql.utils; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.lang.StringUtils; + +/** + * mysql collation转换mapping关系表 + * + * @author agapple 2018年11月5日 下午1:01:15 + * @since 1.1.2 + */ +public class CharsetUtil { + + private static final String[] INDEX_TO_CHARSET = new String[2048]; + private static final Map CHARSET_TO_INDEX = new HashMap(); + static { + INDEX_TO_CHARSET[1] = "big5"; + INDEX_TO_CHARSET[84] = "big5"; + + INDEX_TO_CHARSET[3] = "dec8"; + INDEX_TO_CHARSET[69] = "dec8"; + + INDEX_TO_CHARSET[4] = "cp850"; + INDEX_TO_CHARSET[80] = "cp850"; + + INDEX_TO_CHARSET[6] = "hp8"; + INDEX_TO_CHARSET[72] = "hp8"; + + INDEX_TO_CHARSET[7] = "koi8r"; + INDEX_TO_CHARSET[74] = "koi8r"; + + INDEX_TO_CHARSET[5] = "latin1"; + INDEX_TO_CHARSET[8] = "latin1"; + INDEX_TO_CHARSET[15] = "latin1"; + INDEX_TO_CHARSET[31] = "latin1"; + INDEX_TO_CHARSET[47] = "latin1"; + INDEX_TO_CHARSET[48] = "latin1"; + INDEX_TO_CHARSET[49] = "latin1"; + INDEX_TO_CHARSET[94] = "latin1"; + + INDEX_TO_CHARSET[9] = "latin2"; + INDEX_TO_CHARSET[21] = "latin2"; + INDEX_TO_CHARSET[27] = "latin2"; + INDEX_TO_CHARSET[77] = "latin2"; + + INDEX_TO_CHARSET[10] = "swe7"; + INDEX_TO_CHARSET[82] = "swe7"; + + INDEX_TO_CHARSET[11] = "ascii"; + INDEX_TO_CHARSET[65] = "ascii"; + + INDEX_TO_CHARSET[12] = "ujis"; + INDEX_TO_CHARSET[91] = "ujis"; + + INDEX_TO_CHARSET[13] = "sjis"; + INDEX_TO_CHARSET[88] = "sjis"; + + INDEX_TO_CHARSET[16] = "hebrew"; + INDEX_TO_CHARSET[71] = "hebrew"; + + INDEX_TO_CHARSET[18] = "tis620"; + INDEX_TO_CHARSET[69] = "tis620"; + + INDEX_TO_CHARSET[19] = "euckr"; + INDEX_TO_CHARSET[85] = "euckr"; + + INDEX_TO_CHARSET[22] = "koi8u"; + INDEX_TO_CHARSET[75] = "koi8u"; + + INDEX_TO_CHARSET[24] = "gb2312"; + INDEX_TO_CHARSET[86] = "gb2312"; + + INDEX_TO_CHARSET[25] = "greek"; + INDEX_TO_CHARSET[70] = "greek"; + + INDEX_TO_CHARSET[26] = "cp1250"; + INDEX_TO_CHARSET[34] = "cp1250"; + INDEX_TO_CHARSET[44] = "cp1250"; + INDEX_TO_CHARSET[66] = "cp1250"; + INDEX_TO_CHARSET[99] = "cp1250"; + + INDEX_TO_CHARSET[28] = "gbk"; + INDEX_TO_CHARSET[87] = "gbk"; + + INDEX_TO_CHARSET[30] = "latin5"; + INDEX_TO_CHARSET[78] = "latin5"; + + INDEX_TO_CHARSET[32] = "armscii8"; + INDEX_TO_CHARSET[64] = "armscii8"; + + INDEX_TO_CHARSET[33] = "utf8"; + INDEX_TO_CHARSET[83] = "utf8"; + for (int i = 192; i <= 223; i++) { + INDEX_TO_CHARSET[i] = "utf8"; + } + for (int i = 336; i <= 337; i++) { + INDEX_TO_CHARSET[i] = "utf8"; + } + for (int i = 352; i <= 357; i++) { + INDEX_TO_CHARSET[i] = "utf8"; + } + INDEX_TO_CHARSET[368] = "utf8"; + INDEX_TO_CHARSET[2047] = "utf8"; + + INDEX_TO_CHARSET[35] = "ucs2"; + INDEX_TO_CHARSET[90] = "ucs2"; + for (int i = 128; i <= 151; i++) { + INDEX_TO_CHARSET[i] = "ucs2"; + } + INDEX_TO_CHARSET[159] = "ucs2"; + for (int i = 358; i <= 360; i++) { + INDEX_TO_CHARSET[i] = "ucs2"; + } + + INDEX_TO_CHARSET[36] = "cp866"; + INDEX_TO_CHARSET[68] = "cp866"; + + INDEX_TO_CHARSET[37] = "keybcs2"; + INDEX_TO_CHARSET[73] = "keybcs2"; + + INDEX_TO_CHARSET[38] = "macce"; + INDEX_TO_CHARSET[43] = "macce"; + + INDEX_TO_CHARSET[39] = "macroman"; + INDEX_TO_CHARSET[53] = "macroman"; + + INDEX_TO_CHARSET[40] = "cp852"; + INDEX_TO_CHARSET[81] = "cp852"; + + INDEX_TO_CHARSET[20] = "latin7"; + INDEX_TO_CHARSET[41] = "latin7"; + INDEX_TO_CHARSET[42] = "latin7"; + INDEX_TO_CHARSET[79] = "latin7"; + + INDEX_TO_CHARSET[45] = "utf8mb4"; + INDEX_TO_CHARSET[46] = "utf8mb4"; + for (int i = 224; i <= 247; i++) { + INDEX_TO_CHARSET[i] = "utf8mb4"; + } + for (int i = 255; i <= 271; i++) { + INDEX_TO_CHARSET[i] = "utf8mb4"; + } + for (int i = 273; i <= 275; i++) { + INDEX_TO_CHARSET[i] = "utf8mb4"; + } + for (int i = 277; i <= 294; i++) { + INDEX_TO_CHARSET[i] = "utf8mb4"; + } + for (int i = 296; i <= 298; i++) { + INDEX_TO_CHARSET[i] = "utf8mb4"; + } + INDEX_TO_CHARSET[300] = "utf8mb4"; + for (int i = 303; i <= 307; i++) { + INDEX_TO_CHARSET[i] = "utf8mb4"; + } + INDEX_TO_CHARSET[326] = "utf8mb4"; + INDEX_TO_CHARSET[328] = "utf8mb4"; + + INDEX_TO_CHARSET[14] = "cp1251"; + INDEX_TO_CHARSET[23] = "cp1251"; + INDEX_TO_CHARSET[50] = "cp1251"; + INDEX_TO_CHARSET[51] = "cp1251"; + INDEX_TO_CHARSET[52] = "cp1251"; + + INDEX_TO_CHARSET[54] = "utf16"; + INDEX_TO_CHARSET[55] = "utf16"; + for (int i = 101; i <= 124; i++) { + INDEX_TO_CHARSET[i] = "utf16"; + } + INDEX_TO_CHARSET[327] = "utf16"; + + INDEX_TO_CHARSET[56] = "utf16le"; + INDEX_TO_CHARSET[62] = "utf16le"; + + INDEX_TO_CHARSET[57] = "cp1256"; + INDEX_TO_CHARSET[67] = "cp1256"; + + INDEX_TO_CHARSET[29] = "cp1257"; + INDEX_TO_CHARSET[58] = "cp1257"; + INDEX_TO_CHARSET[59] = "cp1257"; + + INDEX_TO_CHARSET[60] = "utf32"; + INDEX_TO_CHARSET[61] = "utf32"; + for (int i = 160; i <= 183; i++) { + INDEX_TO_CHARSET[i] = "utf32"; + } + INDEX_TO_CHARSET[391] = "utf32"; + + INDEX_TO_CHARSET[63] = "binary"; + + INDEX_TO_CHARSET[92] = "geostd8"; + INDEX_TO_CHARSET[93] = "geostd8"; + + INDEX_TO_CHARSET[95] = "cp932"; + INDEX_TO_CHARSET[96] = "cp932"; + + INDEX_TO_CHARSET[97] = "eucjpms"; + INDEX_TO_CHARSET[98] = "eucjpms"; + + for (int i = 248; i <= 250; i++) { + INDEX_TO_CHARSET[i] = "gb18030"; + } + + // charset --> index + for (int i = 0; i < 2048; i++) { + String charset = INDEX_TO_CHARSET[i]; + if (charset != null && CHARSET_TO_INDEX.get(charset) == null) { + CHARSET_TO_INDEX.put(charset, i); + } + } + CHARSET_TO_INDEX.put("iso-8859-1", 14); + CHARSET_TO_INDEX.put("iso_8859_1", 14); + CHARSET_TO_INDEX.put("utf-8", 33); + CHARSET_TO_INDEX.put("utf8mb4", 45); + } + + public static final String getCharset(int index) { + return INDEX_TO_CHARSET[index]; + } + + public static final int getIndex(String charset) { + if (charset == null || charset.length() == 0) { + return 0; + } else { + Integer i = CHARSET_TO_INDEX.get(charset.toLowerCase()); + return (i == null) ? 0 : i.intValue(); + } + } + + /** + * 'utf8' COLLATE 'utf8_general_ci' + * + * @param charset + * @return + */ + public static final String collateCharset(String charset) { + String[] output = StringUtils.split(charset, "COLLATE"); + return output[0].replace('\'', ' ').trim(); + } + + public static String getJavaCharset(String charset) { + if ("utf8".equals(charset)) { + return charset; + } + + if (StringUtils.endsWithIgnoreCase(charset, "utf8mb4")) { + return "utf-8"; + } + + if (StringUtils.endsWithIgnoreCase(charset, "binary")) { + return "iso_8859_1"; + } + + return charset; + } +} diff --git a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java new file mode 100644 index 00000000..09837d09 --- /dev/null +++ b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java @@ -0,0 +1,51 @@ +package com.alibaba.otter.canal.parse.driver.mysql; + +import org.junit.Assert; +import org.junit.Test; + +import com.alibaba.otter.canal.parse.driver.mysql.utils.CharsetUtil; + +public class CharsetUtilTest { + + @Test + public void testLatin1() { + int charsetIndex = 5; + String charset = "latin1"; + Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex))); + } + + @Test + public void testGbk() { + int charsetIndex = 87; + String charset = "gbk"; + Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex))); + } + + @Test + public void testGb2312() { + int charsetIndex = 24; + String charset = "gb2312"; + Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex))); + } + + @Test + public void testUtf8() { + int charsetIndex = 213; + String charset = "utf8"; + Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex))); + } + + @Test + public void testUtf8mb4() { + int charsetIndex = 235; + String charset = "utf8mb4"; + Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex))); + } + + @Test + public void testBinary() { + int charsetIndex = 63; + String charset = "binary"; + Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex))); + } +} diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java index 8b682f5d..703de818 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java @@ -275,6 +275,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement needDmlParse = true; break; case LogEvent.UPDATE_ROWS_EVENT_V1: + case LogEvent.PARTIAL_UPDATE_ROWS_EVENT: case LogEvent.UPDATE_ROWS_EVENT: tableMeta = logEventConvert.parseRowsEventForTableMeta((UpdateRowsLogEvent) logEvent); needDmlParse = true; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 0b88af14..5c5432c6 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -120,6 +120,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar 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: @@ -261,7 +262,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } boolean isDml = (type == EventType.INSERT || type == EventType.UPDATE || type == EventType.DELETE); - + if (!isSeek && !isDml) { // 使用新的表结构元数据管理方式 EntryPosition position = createPosition(event.getHeader()); @@ -504,7 +505,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar 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) { + } 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; @@ -523,7 +525,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar boolean tableError = false; int rowsCount = 0; - while (buffer.nextOneRow(columns)) { + while (buffer.nextOneRow(columns, false)) { // 处理row记录 RowData.Builder rowDataBuilder = RowData.newBuilder(); if (EventType.INSERT == eventType) { @@ -535,7 +537,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar } else { // update需要处理before/after tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta); - if (!buffer.nextOneRow(changeColumns)) { + if (!buffer.nextOneRow(changeColumns, true)) { rowChangeBuider.addRowDatas(rowDataBuilder.build()); break; } @@ -577,7 +579,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar 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; @@ -631,22 +634,43 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar continue; } - if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { - // 不解析最后一列 - buffer.nextValue(info.type, info.meta, false); - continue; - } - - Column.Builder columnBuilder = Column.newBuilder(); - FieldMeta fieldMeta = null; if (tableMeta != null && !tableError) { // 处理file meta fieldMeta = tableMeta.getFields().get(i); + } + + if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { + // 不解析最后一列 + buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, false); + continue; + } + + if (fieldMeta != null && existOptionalMetaData) { + // 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 , db : " + + fieldMeta.toString() + " , binlog : " + info.toString() + + " , on : " + event.getHeader().getLogFileName() + ":" + + event.getHeader().getLogPos()); + } + } + + 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); @@ -664,12 +688,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar isSingleBit = true; } } - buffer.nextValue(info.type, info.meta, isBinary); - if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) { - // 不解析最后一列 - continue; - } + buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, isBinary); int javaType = buffer.getJavaType(); if (buffer.isNull()) { columnBuilder.setIsNull(true); @@ -683,7 +703,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar case Types.BIGINT: // 处理unsigned类型 Number number = (Number) value; - if (fieldMeta != null && fieldMeta.isUnsigned() && number.longValue() < 0) { + 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 @@ -772,7 +793,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar default: columnBuilder.setValue(value.toString()); } - } columnBuilder.setSqlType(javaType); diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java index 2cb6ed8b..fd01e6a8 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java @@ -1,7 +1,11 @@ package com.alibaba.otter.canal.parse; import java.io.IOException; +import java.io.Serializable; +import java.io.UnsupportedEncodingException; import java.net.InetSocketAddress; +import java.nio.charset.Charset; +import java.util.BitSet; import java.util.List; import org.apache.commons.lang.StringUtils; @@ -24,12 +28,25 @@ import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher; import com.taobao.tddl.dbsync.binlog.LogContext; import com.taobao.tddl.dbsync.binlog.LogDecoder; import com.taobao.tddl.dbsync.binlog.LogEvent; +import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent; import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent; +import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent; import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent; +import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer; +import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent; +import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; +import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo; +import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent; +import com.taobao.tddl.dbsync.binlog.event.XidLogEvent; +import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent; public class DirectLogFetcherTest { protected final Logger logger = LoggerFactory.getLogger(this.getClass()); + protected String binlogFileName = "mysql-bin.000001"; + protected Charset charset = Charset.forName("utf-8"); private boolean isMariaDB; private int binlogChecksum; @@ -37,7 +54,7 @@ public class DirectLogFetcherTest { public void testSimple() { DirectLogFetcher fetcher = new DirectLogFetcher(); try { - MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxx", "xxxx"); + MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello"); connector.connect(); updateSettings(connector); loadBinlogChecksum(connector); @@ -67,21 +84,22 @@ public class DirectLogFetcherTest { break; case LogEvent.WRITE_ROWS_EVENT_V1: case LogEvent.WRITE_ROWS_EVENT: - // parseRowsEvent((WriteRowsLogEvent) event); + parseRowsEvent((WriteRowsLogEvent) event); break; case LogEvent.UPDATE_ROWS_EVENT_V1: + case LogEvent.PARTIAL_UPDATE_ROWS_EVENT: case LogEvent.UPDATE_ROWS_EVENT: - // parseRowsEvent((UpdateRowsLogEvent) event); + parseRowsEvent((UpdateRowsLogEvent) event); break; case LogEvent.DELETE_ROWS_EVENT_V1: case LogEvent.DELETE_ROWS_EVENT: - // parseRowsEvent((DeleteRowsLogEvent) event); + parseRowsEvent((DeleteRowsLogEvent) event); break; case LogEvent.QUERY_EVENT: - // parseQueryEvent((QueryLogEvent) event); + parseQueryEvent((QueryLogEvent) event); break; case LogEvent.ROWS_QUERY_LOG_EVENT: - // parseRowsQueryEvent((RowsQueryLogEvent) event); + parseRowsQueryEvent((RowsQueryLogEvent) event); break; case LogEvent.ANNOTATE_ROWS_EVENT: break; @@ -241,4 +259,98 @@ public class DirectLogFetcherTest { exector.update(cmd); } + protected void parseQueryEvent(QueryLogEvent event) { + System.out.println(String.format("================> binlog[%s:%s] , name[%s]", + binlogFileName, + event.getHeader().getLogPos() - event.getHeader().getEventLen(), + event.getCatalog())); + System.out.println("sql : " + event.getQuery()); + } + + protected void parseRowsQueryEvent(RowsQueryLogEvent event) throws Exception { + System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader() + .getLogPos() - event.getHeader().getEventLen())); + System.out.println("sql : " + new String(event.getRowsQuery().getBytes("ISO-8859-1"), charset.name())); + } + + protected void parseAnnotateRowsEvent(AnnotateRowsEvent event) throws Exception { + System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader() + .getLogPos() - event.getHeader().getEventLen())); + System.out.println("sql : " + new String(event.getRowsQuery().getBytes("ISO-8859-1"), charset.name())); + } + + protected void parseXidEvent(XidLogEvent event) throws Exception { + System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader() + .getLogPos() - event.getHeader().getEventLen())); + System.out.println("xid : " + event.getXid()); + } + + protected void parseRowsEvent(RowsLogEvent event) { + try { + System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s]", + binlogFileName, + event.getHeader().getLogPos() - event.getHeader().getEventLen(), + event.getTable().getDbName(), + event.getTable().getTableName())); + RowsLogBuffer buffer = event.getRowsBuf(charset.name()); + BitSet columns = event.getColumns(); + BitSet changeColumns = event.getChangeColumns(); + while (buffer.nextOneRow(columns)) { + // 处理row记录 + int type = event.getHeader().getType(); + if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) { + // insert的记录放在before字段中 + parseOneRow(event, buffer, columns, true); + } else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) { + // delete的记录放在before字段中 + parseOneRow(event, buffer, columns, false); + } else { + // update需要处理before/after + System.out.println("-------> before"); + parseOneRow(event, buffer, columns, false); + if (!buffer.nextOneRow(changeColumns, true)) { + break; + } + System.out.println("-------> after"); + parseOneRow(event, buffer, changeColumns, true); + } + + } + } catch (Exception e) { + throw new RuntimeException("parse row data failed.", e); + } + } + + protected void parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter) + throws UnsupportedEncodingException { + TableMapLogEvent map = event.getTable(); + if (map == null) { + throw new RuntimeException("not found TableMap with tid=" + event.getTableId()); + } + + final int columnCnt = map.getColumnCnt(); + final ColumnInfo[] columnInfo = map.getColumnInfo(); + + for (int i = 0; i < columnCnt; i++) { + if (!cols.get(i)) { + continue; + } + + ColumnInfo info = columnInfo[i]; + buffer.nextValue(null, i, info.type, info.meta); + + if (buffer.isNull()) { + // + } else { + final Serializable value = buffer.getValue(); + if (value instanceof byte[]) { + System.out.println(new String((byte[]) value)); + } else { + System.out.println(value); + } + } + } + + } + } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java index 79b88375..70b2dbd7 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java @@ -89,6 +89,7 @@ public class MysqlBinlogParsePerformanceTest { parseRowsEvent((WriteRowsLogEvent) event, sum); break; case LogEvent.UPDATE_ROWS_EVENT_V1: + case LogEvent.PARTIAL_UPDATE_ROWS_EVENT: case LogEvent.UPDATE_ROWS_EVENT: parseRowsEvent((UpdateRowsLogEvent) event, sum); break; @@ -154,7 +155,7 @@ public class MysqlBinlogParsePerformanceTest { parseOneRow(event, buffer, columns, false); } else { parseOneRow(event, buffer, columns, false); - if (!buffer.nextOneRow(changeColumns)) { + if (!buffer.nextOneRow(changeColumns, true)) { break; } parseOneRow(event, buffer, changeColumns, true); @@ -182,7 +183,7 @@ public class MysqlBinlogParsePerformanceTest { } ColumnInfo info = columnInfo[i]; - buffer.nextValue(info.type, info.meta); + buffer.nextValue(null, i, info.type, info.meta); if (buffer.isNull()) { } else { buffer.getValue();