From cfa90af33c38435659e7143474a58dad76c5fbe0 Mon Sep 17 00:00:00 2001 From: agapple Date: Mon, 12 Feb 2018 11:01:53 +0800 Subject: [PATCH] fixed unique and upgrade druid --- .../example/AbstractCanalClientTest.java | 15 ++++++++----- .../otter/canal/parse/inbound/TableMeta.java | 21 +++++++++++++++---- .../inbound/mysql/dbsync/TableMetaCache.java | 1 + .../inbound/mysql/tsdb/DatabaseTableMeta.java | 4 ++++ .../inbound/mysql/tsdb/MemoryTableMeta.java | 12 +++++++++++ pom.xml | 2 +- 6 files changed, 45 insertions(+), 10 deletions(-) diff --git a/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java index 59a378a0..60ed8744 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/AbstractCanalClientTest.java @@ -56,10 +56,10 @@ public class AbstractCanalClientTest { context_format += "****************************************************" + SEP; row_format = SEP - + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {} , delay : {}ms" + + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {}({}) , delay : {}ms" + SEP; - transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {} , delay : {}ms" + SEP; + transaction_format = SEP + "================> binlog[{}:{}] , executeTime : {}({}) , delay : {}ms" + SEP; } @@ -165,6 +165,8 @@ public class AbstractCanalClientTest { for (Entry entry : entrys) { long executeTime = entry.getHeader().getExecuteTime(); long delayTime = new Date().getTime() - executeTime; + Date date = new Date(entry.getHeader().getExecuteTime()); + SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN) { @@ -178,7 +180,8 @@ public class AbstractCanalClientTest { logger.info(transaction_format, new Object[] { entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset()), - String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime) }); + String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), + String.valueOf(delayTime) }); logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId()); } else if (entry.getEntryType() == EntryType.TRANSACTIONEND) { TransactionEnd end = null; @@ -193,7 +196,8 @@ public class AbstractCanalClientTest { logger.info(transaction_format, new Object[] { entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset()), - String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime) }); + String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), + String.valueOf(delayTime) }); } continue; @@ -213,7 +217,8 @@ public class AbstractCanalClientTest { new Object[] { entry.getHeader().getLogfileName(), String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType, - String.valueOf(entry.getHeader().getExecuteTime()), String.valueOf(delayTime) }); + String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date), + String.valueOf(delayTime) }); if (eventType == EventType.QUERY || rowChage.getIsDdl()) { logger.info(" sql ----> " + rowChage.getSql() + SEP); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java index 7f56e440..68915ae4 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/TableMeta.java @@ -3,9 +3,10 @@ package com.alibaba.otter.canal.parse.inbound; import java.util.ArrayList; import java.util.List; -import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; import org.apache.commons.lang.StringUtils; +import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent; + /** * 描述数据meta对象,mysql binlog中对应的{@linkplain TableMapLogEvent}包含的信息不全 * @@ -127,6 +128,7 @@ public class TableMeta { private boolean key; private String defaultValue; private String extra; + private boolean unique; public String getColumnName() { return columnName; @@ -180,10 +182,21 @@ public class TableMeta { this.extra = extra; } - public String toString() { - return "FieldMeta [columnName=" + columnName + ", columnType=" + columnType + ", defaultValue=" - + defaultValue + ", nullable=" + nullable + ", key=" + key + "]"; + public boolean isUnique() { + return unique; } + + public void setUnique(boolean unique) { + this.unique = unique; + } + + @Override + public String toString() { + return "FieldMeta [columnName=" + columnName + ", columnType=" + columnType + ", nullable=" + nullable + + ", key=" + key + ", defaultValue=" + defaultValue + ", extra=" + extra + ", unique=" + unique + + "]"; + } + } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java index cc1e5797..8493cb36 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java @@ -104,6 +104,7 @@ public class TableMetaCache { * size), "YES")); meta.setKey("PRI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size))); + meta.setUnique("UNI".equalsIgnoreCase(packet.getFieldValues().get(nameMaps.get(COLUMN_KEY) + i * size))); // 特殊处理引号 meta.setDefaultValue(DruidDdlParser.unescapeQuotaName(packet.getFieldValues() .get(nameMaps.get(COLUMN_DEFAULT) + i * size))); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java index 8c57b3ac..4ddf14ba 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java @@ -446,6 +446,10 @@ public class DatabaseTableMeta implements TableMetaTSDB { if (sourceField.isKey() != targetField.isKey()) { return false; } + + if (sourceField.isUnique() != targetField.isUnique()) { + return false; + } } return true; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java index f36aa726..2cfa1e2f 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java @@ -21,12 +21,14 @@ import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr; import com.alibaba.druid.sql.ast.statement.SQLColumnConstraint; import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition; import com.alibaba.druid.sql.ast.statement.SQLColumnPrimaryKey; +import com.alibaba.druid.sql.ast.statement.SQLColumnUniqueKey; import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement; import com.alibaba.druid.sql.ast.statement.SQLNotNullConstraint; import com.alibaba.druid.sql.ast.statement.SQLNullConstraint; import com.alibaba.druid.sql.ast.statement.SQLSelectOrderByItem; import com.alibaba.druid.sql.ast.statement.SQLTableElement; import com.alibaba.druid.sql.dialect.mysql.ast.MySqlPrimaryKey; +import com.alibaba.druid.sql.dialect.mysql.ast.MySqlUnique; import com.alibaba.druid.sql.repository.Schema; import com.alibaba.druid.sql.repository.SchemaObject; import com.alibaba.druid.sql.repository.SchemaRepository; @@ -204,6 +206,8 @@ public class MemoryTableMeta implements TableMetaTSDB { fieldMeta.setNullable(true); } else if (constraint instanceof SQLColumnPrimaryKey) { fieldMeta.setKey(true); + } else if (constraint instanceof SQLColumnUniqueKey) { + fieldMeta.setUnique(true); } } tableMeta.addFieldMeta(fieldMeta); @@ -215,6 +219,14 @@ public class MemoryTableMeta implements TableMetaTSDB { FieldMeta field = tableMeta.getFieldMetaByName(name); field.setKey(true); } + } else if (element instanceof MySqlUnique) { + MySqlUnique column = (MySqlUnique) element; + List uks = column.getColumns(); + for (SQLSelectOrderByItem uk : uks) { + String name = getSqlName(uk.getExpr()); + FieldMeta field = tableMeta.getFieldMetaByName(name); + field.setUnique(true); + } } } diff --git a/pom.xml b/pom.xml index a951f42c..e28e2806 100644 --- a/pom.xml +++ b/pom.xml @@ -254,7 +254,7 @@ com.alibaba druid - 1.1.7-preview_0 + 1.1.8