feature: support polardb-x 2.0

changes/en-us/2.0.0.md

@@ -25,7 +25,8 @@ The version is updated as follows:
 - [[#4033](https://github.com/seata/seata/pull/4033)] add SQLServer support for Server DB storage mode
 - [[#5717](https://github.com/seata/seata/pull/5717)] compatible with file.conf and registry.conf configurations in version 1.4.2 and below
 - [[#5842](https://github.com/seata/seata/pull/5842)] adding metainfo to docker image
-- 
+- [[#5907](https://github.com/seata/seata/pull/5907)] support polardb-x 2.0 in AT mode
+
 ### bugfix：
 - [[#5677](https://github.com/seata/seata/pull/5677)] fix saga mode serviceTask inputParams json autoType convert exception
 - [[#5194](https://github.com/seata/seata/pull/5194)] fix wrong keyword order for oracle when creating a table

changes/zh-cn/2.0.0.md

@@ -25,6 +25,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
 - [[#4033](https://github.com/seata/seata/pull/4033)] 增加ServerDB存储模式的SQLServer支持
 - [[#5717](https://github.com/seata/seata/pull/5717)] 兼容1.4.2及以下版本的file.conf/registry.conf配置
 - [[#5842](https://github.com/seata/seata/pull/5842)] 构建docker 镜像时添加相关git信息,方便定位代码关系
+- [[#5907](https://github.com/seata/seata/pull/5907)] 增加AT模式的PolarDB-X 2.0数据库支持
 
 ### bugfix：
 - [[#5677](https://github.com/seata/seata/pull/5677)]  修复saga模式下serviceTask入参autoType转化失败问题

common/src/main/java/io/seata/common/util/PageUtil.java

@@ -102,6 +102,7 @@ public static void checkParam(int pageNum, int pageSize) {
     public static String pageSql(String sourceSql, String dbType, int pageNum, int pageSize) {
         switch (dbType) {
             case "mysql":
+            case "polardb-x":
             case "h2":
             case "postgresql":
             case "oceanbase":

common/src/test/java/io/seata/common/util/PageUtilTest.java

@@ -39,6 +39,7 @@ public void testPageSql() {
                 " where rn between 1 and 5";
 
         assertEquals(PageUtil.pageSql(sourceSql, "mysql", 1, 5), mysqlTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "polardb-x", 1, 5), mysqlTargetSql);
         assertEquals(PageUtil.pageSql(sourceSql, "h2", 1, 5), mysqlTargetSql);
         assertEquals(PageUtil.pageSql(sourceSql, "postgresql", 1, 5), mysqlTargetSql);
         assertEquals(PageUtil.pageSql(sourceSql, "oceanbase", 1, 5), mysqlTargetSql);

core/src/main/java/io/seata/core/store/db/sql/lock/PolarDBXLockStoreSql.java

@@ -0,0 +1,27 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.core.store.db.sql.lock;
+
+import io.seata.common.loader.LoadLevel;
+
+/**
+ * Database lock store for PolarDB-X
+ *
+ * @author hsien999
+ */
+@LoadLevel(name = "polardb-x")
+public class PolarDBXLockStoreSql extends MysqlLockStoreSql {
+}

core/src/main/java/io/seata/core/store/db/sql/log/PolarDBXLogStoreSqls.java

@@ -0,0 +1,27 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.core.store.db.sql.log;
+
+import io.seata.common.loader.LoadLevel;
+
+/**
+ * Database log store for PolarDB-X
+ *
+ * @author hsien999
+ */
+@LoadLevel(name = "polardb-x")
+public class PolarDBXLogStoreSqls extends MysqlLogStoreSqls {
+}

core/src/main/resources/META-INF/services/io.seata.core.store.db.sql.lock.LockStoreSql

@@ -4,4 +4,5 @@ io.seata.core.store.db.sql.lock.OceanbaseLockStoreSql
 io.seata.core.store.db.sql.lock.PostgresqlLockStoreSql
 io.seata.core.store.db.sql.lock.H2LockStoreSql
 io.seata.core.store.db.sql.lock.SqlServerLockStoreSql
-io.seata.core.store.db.sql.lock.MariadbLockStoreSql
+io.seata.core.store.db.sql.lock.MariadbLockStoreSql
+io.seata.core.store.db.sql.lock.PolarDBXLockStoreSql

core/src/main/resources/META-INF/services/io.seata.core.store.db.sql.log.LogStoreSqls

@@ -4,4 +4,5 @@ io.seata.core.store.db.sql.log.PostgresqlLogStoreSqls
 io.seata.core.store.db.sql.log.OceanbaseLogStoreSqls
 io.seata.core.store.db.sql.log.H2LogStoreSqls
 io.seata.core.store.db.sql.log.SqlServerLogStoreSqls
-io.seata.core.store.db.sql.log.MariadbLogStoreSqls
+io.seata.core.store.db.sql.log.MariadbLogStoreSqls
+io.seata.core.store.db.sql.log.PolarDBXLogStoreSqls

rm-datasource/src/main/java/io/seata/rm/datasource/DataSourceProxy.java

@@ -19,21 +19,20 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 
 import javax.sql.DataSource;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import io.seata.common.Constants;
-import io.seata.core.constants.DBType;
 import io.seata.core.context.RootContext;
 import io.seata.core.model.BranchType;
 import io.seata.core.model.Resource;
 import io.seata.rm.DefaultResourceManager;
 import io.seata.rm.datasource.sql.struct.TableMetaCacheFactory;
 import io.seata.rm.datasource.util.JdbcUtils;
 import io.seata.sqlparser.util.JdbcConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * The type Data source proxy.
@@ -89,6 +88,8 @@ private void init(DataSource dataSource, String resourceGroupId) {
             dbType = JdbcUtils.getDbType(jdbcUrl);
             if (JdbcConstants.ORACLE.equals(dbType)) {
                 userName = connection.getMetaData().getUserName();
+            } else if (JdbcConstants.MYSQL.equals(dbType)) {
+                getMySQLAdaptiveType(connection);
             }
             version = selectDbVersion(connection);
         } catch (SQLException e) {
@@ -105,6 +106,20 @@ private void init(DataSource dataSource, String resourceGroupId) {
         RootContext.setDefaultBranchType(this.getBranchType());
     }
 
+    /**
+     * get mysql adaptive type for PolarDB-X
+     *
+     * @param connection db connection
+     */
+    private void getMySQLAdaptiveType(Connection connection) {
+        try (Statement statement = connection.createStatement()) {
+            statement.executeQuery("show rule");
+            dbType = JdbcConstants.POLARDBX;
+        } catch (SQLException e) {
+            dbType = JdbcConstants.MYSQL;
+        }
+    }
+
     /**
      * publish tableMeta refresh event
      */
@@ -161,7 +176,7 @@ private void initResourceId() {
             initPGResourceId();
         } else if (JdbcConstants.ORACLE.equals(dbType) && userName != null) {
             initOracleResourceId();
-        } else if (JdbcConstants.MYSQL.equals(dbType)) {
+        } else if (JdbcConstants.MYSQL.equals(dbType) || JdbcConstants.POLARDBX.equals(dbType)) {
             initMysqlResourceId();
         } else if (JdbcConstants.SQLSERVER.equals(dbType)) {
             initSqlServerResourceId();
@@ -293,11 +308,17 @@ public String getVersion() {
     }
 
     private String selectDbVersion(Connection connection) {
-        if (DBType.MYSQL.name().equalsIgnoreCase(dbType)) {
+        if (JdbcConstants.MYSQL.equals(dbType) || JdbcConstants.POLARDBX.equals(dbType)) {
             try (PreparedStatement preparedStatement = connection.prepareStatement("SELECT VERSION()");
                  ResultSet versionResult = preparedStatement.executeQuery()) {
                 if (versionResult.next()) {
-                    return versionResult.getString("VERSION()");
+                    String version = versionResult.getString("VERSION()");
+                    if (version == null) {
+                        return null;
+                    }
+                    int dashIdx = version.indexOf('-');
+                    // in mysql: 5.6.45, in polardb-x: 5.6.45-TDDL-xxx
+                    return dashIdx > 0 ? version.substring(0, dashIdx) : version;
                 }
             } catch (Exception e) {
                 LOGGER.error("get mysql version fail error: {}", e.getMessage());

rm-datasource/src/main/java/io/seata/rm/datasource/exec/ExecuteTemplate.java

@@ -29,6 +29,8 @@
 import io.seata.rm.datasource.exec.mariadb.MariadbUpdateJoinExecutor;
 import io.seata.rm.datasource.exec.mysql.MySQLInsertOnDuplicateUpdateExecutor;
 import io.seata.rm.datasource.exec.mysql.MySQLUpdateJoinExecutor;
+import io.seata.rm.datasource.exec.polardbx.PolarDBXInsertOnDuplicateUpdateExecutor;
+import io.seata.rm.datasource.exec.polardbx.PolarDBXUpdateJoinExecutor;
 import io.seata.rm.datasource.exec.sqlserver.SqlServerDeleteExecutor;
 import io.seata.rm.datasource.exec.sqlserver.SqlServerSelectForUpdateExecutor;
 import io.seata.rm.datasource.exec.sqlserver.SqlServerUpdateExecutor;
@@ -125,11 +127,14 @@ public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecogniz
                         switch (dbType) {
                             case JdbcConstants.MYSQL:
                                 executor =
-                                    new MySQLInsertOnDuplicateUpdateExecutor(statementProxy, statementCallback, sqlRecognizer);
+                                        new MySQLInsertOnDuplicateUpdateExecutor(statementProxy, statementCallback, sqlRecognizer);
                                 break;
                             case JdbcConstants.MARIADB:
                                 executor =
-                                    new MariadbInsertOnDuplicateUpdateExecutor(statementProxy, statementCallback, sqlRecognizer);
+                                        new MariadbInsertOnDuplicateUpdateExecutor(statementProxy, statementCallback, sqlRecognizer);
+                                break;
+                            case JdbcConstants.POLARDBX:
+                                executor = new PolarDBXInsertOnDuplicateUpdateExecutor(statementProxy, statementCallback, sqlRecognizer);
                                 break;
                             default:
                                 throw new NotSupportYetException(dbType + " not support to INSERT_ON_DUPLICATE_UPDATE");
@@ -138,10 +143,13 @@ public static <T, S extends Statement> T execute(List<SQLRecognizer> sqlRecogniz
                     case UPDATE_JOIN:
                         switch (dbType) {
                             case JdbcConstants.MYSQL:
-                                executor = new MySQLUpdateJoinExecutor<>(statementProxy,statementCallback,sqlRecognizer);
+                                executor = new MySQLUpdateJoinExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                                 break;
                             case JdbcConstants.MARIADB:
-                                executor = new MariadbUpdateJoinExecutor<>(statementProxy,statementCallback,sqlRecognizer);
+                                executor = new MariadbUpdateJoinExecutor<>(statementProxy, statementCallback, sqlRecognizer);
+                                break;
+                            case JdbcConstants.POLARDBX:
+                                executor = new PolarDBXUpdateJoinExecutor<>(statementProxy, statementCallback, sqlRecognizer);
                                 break;
                             default:
                                 throw new NotSupportYetException(dbType + " not support to " + SQLType.UPDATE_JOIN.name());

rm-datasource/src/main/java/io/seata/rm/datasource/exec/polardbx/PolarDBXInsertExecutor.java

@@ -0,0 +1,37 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.rm.datasource.exec.polardbx;
+
+import io.seata.common.loader.LoadLevel;
+import io.seata.common.loader.Scope;
+import io.seata.rm.datasource.StatementProxy;
+import io.seata.rm.datasource.exec.StatementCallback;
+import io.seata.rm.datasource.exec.mysql.MySQLInsertExecutor;
+import io.seata.sqlparser.SQLRecognizer;
+import io.seata.sqlparser.util.JdbcConstants;
+
+/**
+ * Insert executor for PolarDB-X
+ *
+ * @author hsien999
+ */
+@LoadLevel(name = JdbcConstants.POLARDBX, scope = Scope.PROTOTYPE)
+public class PolarDBXInsertExecutor extends MySQLInsertExecutor {
+    public PolarDBXInsertExecutor(StatementProxy statementProxy, StatementCallback statementCallback,
+                                  SQLRecognizer sqlRecognizer) {
+        super(statementProxy, statementCallback, sqlRecognizer);
+    }
+}

rm-datasource/src/main/java/io/seata/rm/datasource/exec/polardbx/PolarDBXInsertOnDuplicateUpdateExecutor.java

@@ -0,0 +1,37 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.rm.datasource.exec.polardbx;
+
+import io.seata.common.loader.LoadLevel;
+import io.seata.common.loader.Scope;
+import io.seata.rm.datasource.StatementProxy;
+import io.seata.rm.datasource.exec.StatementCallback;
+import io.seata.rm.datasource.exec.mysql.MySQLInsertOnDuplicateUpdateExecutor;
+import io.seata.sqlparser.SQLRecognizer;
+import io.seata.sqlparser.util.JdbcConstants;
+
+/**
+ * Insert on duplicated update executor for PolarDB-X
+ *
+ * @author hsien999
+ */
+@LoadLevel(name = JdbcConstants.POLARDBX, scope = Scope.PROTOTYPE)
+public class PolarDBXInsertOnDuplicateUpdateExecutor extends MySQLInsertOnDuplicateUpdateExecutor {
+    public PolarDBXInsertOnDuplicateUpdateExecutor(StatementProxy statementProxy, StatementCallback statementCallback,
+                                                   SQLRecognizer sqlRecognizer) {
+        super(statementProxy, statementCallback, sqlRecognizer);
+    }
+}

rm-datasource/src/main/java/io/seata/rm/datasource/exec/polardbx/PolarDBXUpdateJoinExecutor.java

@@ -0,0 +1,36 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.rm.datasource.exec.polardbx;
+
+import java.sql.Statement;
+
+import io.seata.rm.datasource.StatementProxy;
+import io.seata.rm.datasource.exec.StatementCallback;
+import io.seata.rm.datasource.exec.mysql.MySQLUpdateJoinExecutor;
+import io.seata.sqlparser.SQLRecognizer;
+
+/**
+ * Update join executor for PolarDB-X
+ *
+ * @author hsien999
+ */
+public class PolarDBXUpdateJoinExecutor<T, S extends Statement> extends MySQLUpdateJoinExecutor<T, S> {
+    public PolarDBXUpdateJoinExecutor(StatementProxy<S> statementProxy, StatementCallback<T, S> statementCallback,
+                                      SQLRecognizer sqlRecognizer) {
+        super(statementProxy, statementCallback, sqlRecognizer);
+        this.isLowerSupportGroupByPksVersion = false;
+    }
+}

rm-datasource/src/main/java/io/seata/rm/datasource/sql/handler/polardbx/PolarDBXEscapeHandler.java

@@ -0,0 +1,29 @@
+/*
+ *  Copyright 1999-2019 Seata.io Group.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+package io.seata.rm.datasource.sql.handler.polardbx;
+
+import io.seata.common.loader.LoadLevel;
+import io.seata.rm.datasource.sql.handler.mysql.MySQLEscapeHandler;
+import io.seata.sqlparser.util.JdbcConstants;
+
+/**
+ * Escape handler for PolarDB-X
+ *
+ * @author hsien999
+ */
+@LoadLevel(name = JdbcConstants.POLARDBX)
+public class PolarDBXEscapeHandler extends MySQLEscapeHandler {
+}