chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix issue #751

Browse Source
This commit is contained in:
winger
2018-07-17 12:04:27 +08:00
parent bef6a66eaf
commit 441d2d2449
1 changed files with 16 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
View File

@@ -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();
}