From 441d2d244929f6b28aacdf1c6c58a749cbe87c60 Mon Sep 17 00:00:00 2001 From: winger Date: Tue, 17 Jul 2018 12:04:27 +0800 Subject: [PATCH] fix issue #751 --- .../inbound/mysql/dbsync/LogEventConvert.java | 19 ++++++++++++++++--- 1 file changed, 16 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 f99d551c..153a0ecd 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 @@ -504,9 +504,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar throw new CanalParseException("unsupport event type :" + event.getHeader().getType()); } - TableMapLogEvent table = event.getTable(); - Header header = createHeader(event.getHeader(), table.getDbName(), table.getTableName(), eventType); - RowChange.Builder rowChangeBuider = RowChange.newBuilder(); rowChangeBuider.setTableId(event.getTableId()); rowChangeBuider.setIsDdl(false); @@ -517,6 +514,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar BitSet changeColumns = event.getChangeColumns(); boolean tableError = false; + int rowsCount = 0; while (buffer.nextOneRow(columns)) { // 处理row记录 RowData.Builder rowDataBuilder = RowData.newBuilder(); @@ -537,9 +535,13 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta); } + rowsCount ++; rowChangeBuider.addRowDatas(rowDataBuilder.build()); } + TableMapLogEvent table = event.getTable(); + Header header = createHeader(event.getHeader(), table.getDbName(), table.getTableName(), eventType, rowsCount); + RowChange rowChange = rowChangeBuider.build(); if (tableError) { Entry entry = createEntry(header, EntryType.ROWDATA, ByteString.EMPTY); @@ -793,7 +795,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return createEntry(header, EntryType.ROWDATA, rowChangeBuider.build().toByteString()); } + private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType) { + return createHeader(logHeader, schemaName, tableName, eventType, -1); + } + + private Header createHeader(LogHeader logHeader, String schemaName, String tableName, EventType eventType, Integer rowsCount) { // header会做信息冗余,方便以后做检索或者过滤 Header.Builder headerBuilder = Header.newBuilder(); headerBuilder.setVersion(version); @@ -818,6 +825,12 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar String gtid = gtidSet.toString(); headerBuilder.setGtid(gtid); } + + // add rowsCount suppport + if (rowsCount > 0) { + Pair pair = createSpecialPair("rowsCount", String.valueOf(rowsCount)); + headerBuilder.addProps(pair); + } return headerBuilder.build(); }