fixed drds support

alibaba · Oct 27, 2018 · bf1f1b9 · bf1f1b9
1 parent 0660911
commit bf1f1b9
Show file tree

Hide file tree

Showing 3 changed files with 31 additions and 25 deletions.
diff --git a/node/canal/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml b/node/canal/src/main/resources/spring/tsdb/sql-map/sqlmap_history.xml
@@ -36,10 +36,10 @@
     </delete>
 
 
-    <delete id="deleteByGmtModified" parameterClass="java.util.Map">
+    <delete id="deleteByTimestamp" parameterClass="java.util.Map">
         <![CDATA[
 		delete from meta_history
-		where gmt_modified < timestamp(#timestamp#)
+		where destination=#destination# and binlog_timestamp < #timestamp#
         ]]>
     </delete>
 </sqlMap>
diff --git a/node/canal/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml b/node/canal/src/main/resources/spring/tsdb/sql-map/sqlmap_snapshot.xml
@@ -42,10 +42,10 @@
         where destination=#destination#
     </delete>
 
-    <delete id="deleteByGmtModified" parameterClass="java.util.Map">
+    <delete id="deleteByTimestamp" parameterClass="java.util.Map">
         <![CDATA[
 		delete from meta_snapshot
-		where gmt_modified < timestamp(#timestamp#)
+		where destination=#destination# and binlog_timestamp < #timestamp# and binlog_timestamp > 0
         ]]>
     </delete>
 </sqlMap>
diff --git a/node/etl/src/main/java/com/alibaba/otter/node/etl/select/selector/MessageParser.java b/node/etl/src/main/java/com/alibaba/otter/node/etl/select/selector/MessageParser.java
@@ -47,6 +47,7 @@
 import com.alibaba.otter.shared.common.model.config.data.DataMedia;
 import com.alibaba.otter.shared.common.model.config.data.DataMedia.ModeValue;
 import com.alibaba.otter.shared.common.model.config.data.DataMediaPair;
+import com.alibaba.otter.shared.common.model.config.data.DataMediaSource;
 import com.alibaba.otter.shared.common.model.config.data.db.DbMediaSource;
 import com.alibaba.otter.shared.common.model.config.pipeline.Pipeline;
 import com.alibaba.otter.shared.common.model.config.pipeline.PipelineParameter;
@@ -439,27 +440,32 @@ private EventData internParse(Pipeline pipeline, Entry entry, RowChange rowChang
                 }
                 // 获取一下目标库的拆分字段,设置源表为主键
                 // 首先要求源和目标的库名表名是一致的
-                DbDialect targetDbDialect = dbDialectFactory.getDbDialect(pipeline.getId(),
-                    (DbMediaSource) targetDataMedia.getSource());
-                if (targetDbDialect.isDRDS()) {
-                    String schemaName = buildName(eventData.getSchemaName(),
-                        dataMedia.getNamespaceMode(),
-                        targetDataMedia.getNamespaceMode());
-                    String tableName = buildName(eventData.getSchemaName(),
-                        dataMedia.getNameMode(),
-                        targetDataMedia.getNameMode());
-                    String shardColumns = targetDbDialect.getShardColumns(schemaName, tableName);
-                    if (StringUtils.isNotEmpty(shardColumns)) {
-                        String columns[] = StringUtils.split(shardColumns, ',');
-                        for (String key : columns) {
-                            org.apache.ddlutils.model.Column col = table.findColumn(key, false);
-                            if (col != null) {
-                                col.setPrimaryKey(true);
-                            } else {
-                                logger.warn(String.format("shardColumn %s in table[%s.%s] is not found",
-                                    key,
-                                    eventData.getSchemaName(),
-                                    eventData.getTableName()));
+                DataMediaSource targetSource = targetDataMedia.getSource();
+                if (targetSource instanceof DbMediaSource
+                    && StringUtils.containsIgnoreCase(((DbMediaSource) targetSource).getUrl(), "drds")) {
+                    // 优先判断是否为drds
+                    DbDialect targetDbDialect = dbDialectFactory.getDbDialect(pipeline.getId(),
+                        (DbMediaSource) targetDataMedia.getSource());
+                    if (targetDbDialect.isDRDS()) {
+                        String schemaName = buildName(eventData.getSchemaName(),
+                            dataMedia.getNamespaceMode(),
+                            targetDataMedia.getNamespaceMode());
+                        String tableName = buildName(eventData.getSchemaName(),
+                            dataMedia.getNameMode(),
+                            targetDataMedia.getNameMode());
+                        String shardColumns = targetDbDialect.getShardColumns(schemaName, tableName);
+                        if (StringUtils.isNotEmpty(shardColumns)) {
+                            String columns[] = StringUtils.split(shardColumns, ',');
+                            for (String key : columns) {
+                                org.apache.ddlutils.model.Column col = table.findColumn(key, false);
+                                if (col != null) {
+                                    col.setPrimaryKey(true);
+                                } else {
+                                    logger.warn(String.format("shardColumn %s in table[%s.%s] is not found",
+                                        key,
+                                        eventData.getSchemaName(),
+                                        eventData.getTableName()));
+                                }
                             }
                         }
                     }