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

migrate sql parser from druid to polardbx-parser (#5097)

Browse Source 
* migrate parser from druid to polardbx-parser

* migrate parser from druid to polardbx-parser(with version 5.4.19)

* fix bug : full ddl sqls should not assign to extra field but data field

* fix bug : full ddl sqls should not assign to extra field but data field

---------

Co-authored-by: agapple <jianghang.loujh@alibaba-inc.com>
This commit is contained in:
lulu2panpan
2024-05-25 22:05:06 +08:00
committed by GitHub
parent aa05d8e2a0
commit d66a15fe35
12 changed files with 99 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
common/pom.xml
View File

@@ -87,6 +87,11 @@
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>

6
common/src/main/java/com/alibaba/otter/canal/common/utils/CommonUtils.java
View File

@@ -1,5 +1,7 @@
package com.alibaba.otter.canal.common.utils;
import com.alibaba.druid.filter.config.ConfigTools;
import java.io.File;
/**
@@ -50,4 +52,8 @@ public class CommonUtils {
return dirFile.delete();
}
public static String decryptDruidPassword(String publicKeyText, String cipherText) throws Exception {
return ConfigTools.decrypt(publicKeyText, cipherText);
}
}

4
parse/pom.xml
View File

@@ -46,8 +46,8 @@
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>druid</artifactId>
<groupId>com.alibaba.polardbx</groupId>
<artifactId>polardbx-parser</artifactId>
</dependency>
<dependency>
<groupId>mysql</groupId>

2
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java
View File

@@ -258,7 +258,7 @@ public class TableMetaCache {
.append('`')
.append('.')
.append('`')
.append(table)
.append(StringUtils.replace(table,"`","``"))
.append('`')
.toString();
}

63
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/ddl/DruidDdlParser.java
View File

@@ -4,39 +4,38 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import com.alibaba.polardbx.druid.sql.SQLUtils;
import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableAddIndex;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropIndex;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableDropKey;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableItem;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableRename;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLAlterTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateDatabaseStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateIndexStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDeleteStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropDatabaseStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropIndexStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLDropTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableSource;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTruncateStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLUnique;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLUpdateStatement;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement;
import com.alibaba.polardbx.druid.sql.parser.ParserException;
import com.alibaba.polardbx.druid.util.JdbcConstants;
import org.apache.commons.lang.StringUtils;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableAddConstraint;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableAddIndex;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableDropConstraint;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableDropIndex;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableDropKey;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableItem;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableRename;
import com.alibaba.druid.sql.ast.statement.SQLAlterTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLConstraint;
import com.alibaba.druid.sql.ast.statement.SQLCreateDatabaseStatement;
import com.alibaba.druid.sql.ast.statement.SQLCreateIndexStatement;
import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLDeleteStatement;
import com.alibaba.druid.sql.ast.statement.SQLDropDatabaseStatement;
import com.alibaba.druid.sql.ast.statement.SQLDropIndexStatement;
import com.alibaba.druid.sql.ast.statement.SQLDropTableStatement;
import com.alibaba.druid.sql.ast.statement.SQLExprTableSource;
import com.alibaba.druid.sql.ast.statement.SQLInsertStatement;
import com.alibaba.druid.sql.ast.statement.SQLTableSource;
import com.alibaba.druid.sql.ast.statement.SQLTruncateStatement;
import com.alibaba.druid.sql.ast.statement.SQLUnique;
import com.alibaba.druid.sql.ast.statement.SQLUpdateStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement;
import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlRenameTableStatement.Item;
import com.alibaba.druid.sql.parser.ParserException;
import com.alibaba.druid.util.JdbcConstants;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
/**
@@ -141,7 +140,7 @@ public class DruidDdlParser {
}
} else if (statement instanceof MySqlRenameTableStatement) {
MySqlRenameTableStatement rename = (MySqlRenameTableStatement) statement;
for (Item item : rename.getItems()) {
for (MySqlRenameTableStatement.Item item : rename.getItems()) {
DdlResult ddlResult = new DdlResult();
processName(ddlResult, schmeaName, item.getName(), true);
processName(ddlResult, schmeaName, item.getTo(), false);

11
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
View File

@@ -13,14 +13,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.regex.Pattern;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.FieldPacket;
import com.alibaba.polardbx.druid.sql.repository.Schema;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.lang.ObjectUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import com.alibaba.druid.sql.repository.Schema;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.otter.canal.filter.CanalEventFilter;
@@ -227,7 +228,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
StringBuilder sql = new StringBuilder();
for (String table : tables) {
sql.append("show create table `" + schema + "`.`" + table + "`;");
sql.append("show create table `" + schema + "`.`" + StringUtils.replace(table,"`","``") + "`;");
}
List<ResultSetPacket> packets = connection.queryMulti(sql.toString());
@@ -363,7 +364,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
snapshotDO.setBinlogOffest(position.getPosition());
snapshotDO.setBinlogMasterId(String.valueOf(position.getServerId()));
snapshotDO.setBinlogTimestamp(position.getTimestamp());
snapshotDO.setExtra(JSON.toJSONString(schemaDdls));
snapshotDO.setData(JSON.toJSONString(schemaDdls));
try {
metaSnapshotDAO.insert(snapshotDO);
} catch (Throwable e) {
@@ -508,7 +509,7 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private String getFullName(String schema, String table) {
StringBuilder builder = new StringBuilder();
return builder.append(structureSchema(schema)).append('.').append('`').append(table).append('`').toString();
return builder.append(structureSchema(schema)).append('.').append('`').append(StringUtils.replace(table,"`","``")).append('`').toString();
}
public static boolean compareTableMeta(TableMeta source, TableMeta target) {

50
parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
View File

@@ -3,27 +3,38 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import com.alibaba.polardbx.druid.DbType;
import com.alibaba.polardbx.druid.sql.SQLUtils;
import com.alibaba.polardbx.druid.sql.ast.SQLDataType;
import com.alibaba.polardbx.druid.sql.ast.SQLDataTypeImpl;
import com.alibaba.polardbx.druid.sql.ast.SQLExpr;
import com.alibaba.polardbx.druid.sql.ast.SQLStatement;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLCharExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLIdentifierExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLMethodInvokeExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLNullExpr;
import com.alibaba.polardbx.druid.sql.ast.expr.SQLPropertyExpr;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnDefinition;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnPrimaryKey;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLColumnUniqueKey;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLCreateTableStatement;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLNotNullConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLNullConstraint;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLSelectOrderByItem;
import com.alibaba.polardbx.druid.sql.ast.statement.SQLTableElement;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlPrimaryKey;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.MySqlUnique;
import com.alibaba.polardbx.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
import com.alibaba.polardbx.druid.sql.repository.Schema;
import com.alibaba.polardbx.druid.sql.repository.SchemaObject;
import com.alibaba.polardbx.druid.sql.repository.SchemaRepository;
import com.alibaba.polardbx.druid.sql.visitor.SQLASTOutputVisitor;
import com.alibaba.polardbx.druid.sql.visitor.VisitorFeature;
import com.alibaba.polardbx.druid.util.JdbcConstants;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.druid.DbType;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.ast.SQLDataType;
import com.alibaba.druid.sql.ast.SQLDataTypeImpl;
import com.alibaba.druid.sql.ast.SQLExpr;
import com.alibaba.druid.sql.ast.SQLStatement;
import com.alibaba.druid.sql.ast.expr.*;
import com.alibaba.druid.sql.ast.statement.*;
import com.alibaba.druid.sql.dialect.mysql.ast.MySqlPrimaryKey;
import com.alibaba.druid.sql.dialect.mysql.ast.MySqlUnique;
import com.alibaba.druid.sql.dialect.mysql.ast.expr.MySqlOrderingExpr;
import com.alibaba.druid.sql.repository.Schema;
import com.alibaba.druid.sql.repository.SchemaObject;
import com.alibaba.druid.sql.repository.SchemaRepository;
import com.alibaba.druid.sql.visitor.SQLASTOutputVisitor;
import com.alibaba.druid.sql.visitor.VisitorFeature;
import com.alibaba.druid.util.JdbcConstants;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
import com.alibaba.otter.canal.parse.inbound.TableMeta.FieldMeta;
import com.alibaba.otter.canal.parse.inbound.mysql.ddl.DruidDdlParser;
@@ -39,9 +50,10 @@ public class MemoryTableMeta implements TableMetaTSDB {
private Logger logger = LoggerFactory.getLogger(MemoryTableMeta.class);
private Map<List<String>, TableMeta> tableMetas = new ConcurrentHashMap<>();
private SchemaRepository repository = new SchemaRepository(JdbcConstants.MYSQL);
private SchemaRepository repository;
public MemoryTableMeta(){
repository = new SchemaRepository(JdbcConstants.MYSQL);
}
@Override

4
parse/src/main/java/com/alibaba/otter/canal/parse/support/AuthenticationInfo.java
View File

@@ -2,7 +2,7 @@ package com.alibaba.otter.canal.parse.support;
import java.net.InetSocketAddress;
import com.alibaba.druid.filter.config.ConfigTools;
import com.alibaba.otter.canal.common.utils.CommonUtils;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.lang.builder.ToStringStyle;
@@ -25,7 +25,7 @@ public class AuthenticationInfo {
public void initPwd() throws Exception{
if (enableDruid) {
this.password = ConfigTools.decrypt(pwdPublicKey, password);
this.password = CommonUtils.decryptDruidPassword(pwdPublicKey, password);
}
}

16
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/FastsqlSchemaTest.java
View File

@@ -3,14 +3,14 @@ package com.alibaba.otter.canal.parse.inbound.mysql.tsdb;
import org.junit.Assert;
import org.junit.Test;
import com.alibaba.druid.DbType;
import com.alibaba.druid.sql.SQLUtils;
import com.alibaba.druid.sql.repository.Schema;
import com.alibaba.druid.sql.repository.SchemaObject;
import com.alibaba.druid.sql.repository.SchemaRepository;
import com.alibaba.druid.sql.visitor.SQLASTOutputVisitor;
import com.alibaba.druid.sql.visitor.VisitorFeature;
import com.alibaba.druid.util.JdbcConstants;
import com.alibaba.polardbx.druid.DbType;
import com.alibaba.polardbx.druid.sql.SQLUtils;
import com.alibaba.polardbx.druid.sql.repository.Schema;
import com.alibaba.polardbx.druid.sql.repository.SchemaObject;
import com.alibaba.polardbx.druid.sql.repository.SchemaRepository;
import com.alibaba.polardbx.druid.sql.visitor.SQLASTOutputVisitor;
import com.alibaba.polardbx.druid.sql.visitor.VisitorFeature;
import com.alibaba.polardbx.druid.util.JdbcConstants;
public class FastsqlSchemaTest {

5
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_DDL_Test.java
View File

@@ -6,9 +6,6 @@ import java.net.URL;
import java.util.ArrayList;
import java.util.List;
import com.alibaba.druid.sql.repository.SchemaObject;
import com.alibaba.druid.sql.repository.SchemaRepository;
import com.alibaba.druid.util.JdbcConstants;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang.StringUtils;
import org.junit.Assert;
@@ -17,7 +14,7 @@ import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.alibaba.druid.sql.repository.Schema;
import com.alibaba.polardbx.druid.sql.repository.Schema;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
/**

2
parse/src/test/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta_Random_DDL_Test.java
View File

@@ -14,7 +14,7 @@ import org.junit.runner.RunWith;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.alibaba.druid.sql.repository.Schema;
import com.alibaba.polardbx.druid.sql.repository.Schema;
import com.alibaba.otter.canal.parse.inbound.TableMeta;
/**

5
pom.xml
View File

@@ -389,6 +389,11 @@
<artifactId>log4j-to-slf4j</artifactId>
<version>${log4j_version}</version>
</dependency>
<dependency>
<groupId>com.alibaba.polardbx</groupId>
<artifactId>polardbx-parser</artifactId>
<version>5.4.19</version>
</dependency>
<!-- test dependency -->
<dependency>