bugfix: fix SQLServer-related SQL errors in seata server when using d…

…atabase of SQLServer (#6493)

changes/en-us/2.x.md

@@ -34,6 +34,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#6372](https://github.com/apache/incubator-seata/pull/6372)] fix initializing the sql file postgresql.sql index name conflict
 - [[#6380](https://github.com/apache/incubator-seata/pull/6380)] fix sql exception when checking for the existence of the UNDO_LOG table on SQL server
 - [[#6385](https://github.com/apache/incubator-seata/pull/6385)] fix the bug where Role.participant does not execute hooks but clears them.
+- [[#6493](https://github.com/apache/incubator-seata/pull/6493)] fix SQLServer-related SQL error in seata server when using database of SQLServer
 
 ### optimize:
 - [[#6031](https://github.com/apache/incubator-seata/pull/6031)] add a check for the existence of the undolog table

changes/zh-cn/2.x.md

@@ -34,7 +34,7 @@
 - [[#6372](https://github.com/apache/incubator-seata/pull/6372)] 修复初始化sql文件postgresql.sql 索引名称冲突问题
 - [[#6380](https://github.com/apache/incubator-seata/pull/6380)] 修复针对sql server检查UNDO_LOG表是否存在时的SQL异常
 - [[#6385](https://github.com/apache/incubator-seata/pull/6385)] 修复Role.Participant不执行hook但会清理的问题
-
+- [[#6493](https://github.com/apache/incubator-seata/pull/6493)] 修复当使用数据库为SQLServer时seata server的SQL报错
 
 
 ### optimize:

common/src/main/java/org/apache/seata/common/util/PageUtil.java

@@ -48,6 +48,10 @@ public class PageUtil {
      * The constant SOURCE_SQL_PLACE_HOLD
      */
     private static final String SOURCE_SQL_PLACE_HOLD = " #sourcesql# ";
+    /**
+     * The constant ORDER_BY_PLACE_HOLD
+     */
+    private static final String ORDER_BY_PLACE_HOLD = " #order_by# ";
     /**
      * The constant LIMIT_PLACE_HOLD
      */
@@ -78,7 +82,7 @@ public class PageUtil {
     /**
      * The constant SQLSERVER_PAGE_TEMPLATE
      */
-    private static final String SQLSERVER_PAGE_TEMPLATE = "select * from (select temp.*, ROW_NUMBER() OVER(ORDER BY (select NULL)) AS rowId from ("
+    private static final String SQLSERVER_PAGE_TEMPLATE = "select * from (select temp.*, ROW_NUMBER() OVER(ORDER BY " + ORDER_BY_PLACE_HOLD + ") AS rowId from ("
             + SOURCE_SQL_PLACE_HOLD + ") temp ) t where t.rowId between " + START_PLACE_HOLD + " and " + END_PLACE_HOLD;
     /**
      * check page parm
@@ -104,7 +108,7 @@ public static void checkParam(int pageNum, int pageSize) {
      * @param pageSize the page size
      * @return the page sql
      */
-    public static String pageSql(String sourceSql, String dbType, int pageNum, int pageSize) {
+    public static String pageSql(String sourceSql, String dbType, int pageNum, int pageSize, String orderByCondition) {
         switch (dbType) {
             case "mysql":
             case "h2":
@@ -120,6 +124,7 @@ public static String pageSql(String sourceSql, String dbType, int pageNum, int p
                         .replace(END_PLACE_HOLD, String.valueOf(pageSize * pageNum));
             case "sqlserver":
                 return SQLSERVER_PAGE_TEMPLATE.replace(SOURCE_SQL_PLACE_HOLD, sourceSql)
+                        .replace(ORDER_BY_PLACE_HOLD, orderByCondition)
                         .replace(START_PLACE_HOLD, String.valueOf(pageSize * (pageNum - 1) + 1))
                         .replace(END_PLACE_HOLD, String.valueOf(pageSize * pageNum));
             default:

common/src/test/java/org/apache/seata/common/util/PageUtilTest.java

@@ -37,17 +37,17 @@ public void testPageSql() {
         String oracleTargetSql = "select * from " +
                 "( select ROWNUM rn, temp.* from (select * from test where a = 1) temp )" +
                 " where rn between 1 and 5";
-        String sqlserverTargetSql = "select * from (select temp.*, ROW_NUMBER() OVER(ORDER BY (select NULL)) AS rowId from (select * from test where a = 1) temp ) t where t.rowId between 1 and 5";
+        String sqlserverTargetSql = "select * from (select temp.*, ROW_NUMBER() OVER(ORDER BY gmt_create desc) AS rowId from (select * from test where a = 1) temp ) t where t.rowId between 1 and 5";
 
-        assertEquals(PageUtil.pageSql(sourceSql, "mysql", 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);
-        assertEquals(PageUtil.pageSql(sourceSql, "dm", 1, 5), mysqlTargetSql);
-        assertEquals(PageUtil.pageSql(sourceSql, "oracle", 1, 5), oracleTargetSql);
-        assertEquals(PageUtil.pageSql(sourceSql, "sqlserver", 1, 5), sqlserverTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "mysql", 1, 5, null), mysqlTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "h2", 1, 5, null), mysqlTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "postgresql", 1, 5, null), mysqlTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "oceanbase", 1, 5, null), mysqlTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "dm", 1, 5, null), mysqlTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "oracle", 1, 5, null), oracleTargetSql);
+        assertEquals(PageUtil.pageSql(sourceSql, "sqlserver", 1, 5, "gmt_create desc"), sqlserverTargetSql);
 
-        assertThrows(NotSupportYetException.class, () -> PageUtil.pageSql(sourceSql, "xxx", 1, 5));
+        assertThrows(NotSupportYetException.class, () -> PageUtil.pageSql(sourceSql, "xxx", 1, 5, null));
     }
 
     @Test

core/src/main/java/org/apache/seata/core/store/db/sql/distributed/lock/BaseDistributedLockSqlServer.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.seata.core.store.db.sql.distributed.lock;
+
+import org.apache.seata.core.constants.ServerTableColumnsName;
+
+public class BaseDistributedLockSqlServer extends BaseDistributedLockSql {
+
+    protected static final String SELECT_FOR_UPDATE_SQL = "SELECT " + ALL_COLUMNS + " FROM " + DISTRIBUTED_LOCK_TABLE_PLACE_HOLD
+            + " WITH (ROWLOCK, UPDLOCK, HOLDLOCK) WHERE " + ServerTableColumnsName.DISTRIBUTED_LOCK_KEY + " = ?";
+
+    @Override
+    public String getSelectDistributeForUpdateSql(String distributedLockTable) {
+        return SELECT_FOR_UPDATE_SQL.replace(DISTRIBUTED_LOCK_TABLE_PLACE_HOLD, distributedLockTable);
+    }
+
+}

core/src/main/java/org/apache/seata/core/store/db/sql/distributed/lock/DistributedLockSqlFactory.java

@@ -17,16 +17,22 @@
 package org.apache.seata.core.store.db.sql.distributed.lock;
 
 
+import org.apache.seata.core.constants.DBType;
+
 public class DistributedLockSqlFactory {
     private static final DistributedLockSql DISTRIBUTED_LOCK_SQL = new BaseDistributedLockSql();
+    private static final DistributedLockSql DISTRIBUTED_LOCK_SQL_SERVER = new BaseDistributedLockSqlServer();
 
     /**
      * get the lock store sql
      *
-     * @param dbType the dbType, support mysql/oracle/h2/postgre/oceanbase/dm, it's useless now, but maybe useful later
+     * @param dbType the dbType, support mysql/oracle/h2/postgre/oceanbase/dm/sqlserver ...
      * @return lock store sql
      */
     public static DistributedLockSql getDistributedLogStoreSql(String dbType) {
+        if (DBType.SQLSERVER.name().equalsIgnoreCase(dbType)) {
+            return DISTRIBUTED_LOCK_SQL_SERVER;
+        }
         return DISTRIBUTED_LOCK_SQL;
     }
 }

core/src/main/java/org/apache/seata/core/store/db/sql/lock/SqlServerLockStoreSql.java

@@ -29,10 +29,21 @@ public class SqlServerLockStoreSql extends AbstractLockStoreSql {
      * The constant INSERT_LOCK_SQL_MYSQL.
      */
     private static final String INSERT_LOCK_SQL_SQLSERVER = "insert into " + LOCK_TABLE_PLACE_HOLD + "(" + ALL_COLUMNS + ")"
-            + " values (?, ?, ?, ?, ?, ?, ?, SYSDATETIME(), SYSDATETIME())";
+            + " values (?, ?, ?, ?, ?, ?, ?, SYSDATETIME(), SYSDATETIME(), ?)";
+
+    /**
+     * The constant QUERY_ALL_LOCK.
+     */
+    private static final String QUERY_ALL_LOCK_SQLSERVER = "select " + ALL_COLUMNS + " from " + LOCK_TABLE_PLACE_HOLD
+            + WHERE_PLACE_HOLD;
 
     @Override
     public String getInsertLockSQL(String lockTable) {
         return INSERT_LOCK_SQL_SQLSERVER.replace(LOCK_TABLE_PLACE_HOLD, lockTable);
     }
+
+    @Override
+    public String getAllLockSql(String lockTable, String whereCondition) {
+        return QUERY_ALL_LOCK_SQLSERVER.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(WHERE_PLACE_HOLD, whereCondition);
+    }
 }

core/src/main/java/org/apache/seata/core/store/db/sql/log/SqlServerLogStoreSqls.java

@@ -26,6 +26,12 @@
 @LoadLevel(name = "sqlserver")
 public class SqlServerLogStoreSqls extends AbstractLogStoreSqls {
 
+    /**
+     * The constant QUERY_ALL_GLOBAL_SESSION_SQLSERVER.
+     */
+    private static final String QUERY_ALL_GLOBAL_SESSION_SQLSERVER = "select " + ALL_GLOBAL_COLUMNS + " from "
+            + GLOBAL_TABLE_PLACEHOLD + WHERE_PLACEHOLD;
+
     /**
      * The constant INSERT_GLOBAL_TRANSACTION_SQLSERVER.
      */
@@ -85,6 +91,11 @@ public class SqlServerLogStoreSqls extends AbstractLogStoreSqls {
             + " where " + ServerTableColumnsName.BRANCH_TABLE_XID + " = ?"
             + "   and " + ServerTableColumnsName.BRANCH_TABLE_BRANCH_ID + " = ?";
 
+    @Override
+    public String getAllGlobalSessionSql(String globalTable, String whereCondition) {
+        return QUERY_ALL_GLOBAL_SESSION_SQLSERVER.replace(GLOBAL_TABLE_PLACEHOLD, globalTable).replace(WHERE_PLACEHOLD, whereCondition);
+    }
+
     @Override
     public String getInsertGlobalTransactionSQL(String globalTable) {
         return INSERT_GLOBAL_TRANSACTION_SQLSERVER.replace(GLOBAL_TABLE_PLACEHOLD, globalTable);

script/server/db/sqlserver.sql

@@ -95,3 +95,18 @@ CREATE NONCLUSTERED INDEX [idx_branch_id]
                      [branch_id]
         )
 GO
+
+-- the table to store distributed lock constants
+CREATE TABLE [distributed_lock]
+(
+    [lock_key]   char(20)    not null primary key,
+    [lock_value] varchar(20) not null,
+    [expire]     bigint
+    )
+GO
+
+INSERT INTO [distributed_lock] (lock_key, lock_value, expire) VALUES ('AsyncCommitting', ' ', 0);
+INSERT INTO [distributed_lock] (lock_key, lock_value, expire) VALUES ('RetryCommitting', ' ', 0);
+INSERT INTO [distributed_lock] (lock_key, lock_value, expire) VALUES ('RetryRollbacking', ' ', 0);
+INSERT INTO [distributed_lock] (lock_key, lock_value, expire) VALUES ('TxTimeoutCheck', ' ', 0);
+INSERT INTO [distributed_lock] (lock_key, lock_value, expire) VALUES ('UndologDelete', ' ', 0);

server/src/main/java/org/apache/seata/server/console/impl/db/GlobalLockDBServiceImpl.java

@@ -80,9 +80,10 @@ public PageResult<GlobalLockVO> query(GlobalLockParam param) {
 
         List<Object> sqlParamList = new ArrayList<>();
         String whereCondition = this.getWhereConditionByParam(param, sqlParamList);
+        String orderByCondition = getOrderByCondition();
 
         String sourceSql = LockStoreSqlFactory.getLogStoreSql(dbType).getAllLockSql(lockTable, whereCondition);
-        String queryLockSql = PageUtil.pageSql(sourceSql, dbType, param.getPageNum(), param.getPageSize());
+        String queryLockSql = PageUtil.pageSql(sourceSql, dbType, param.getPageNum(), param.getPageSize(), orderByCondition);
         String lockCountSql = PageUtil.countSql(sourceSql, dbType);
 
         List<GlobalLockVO> list = new ArrayList<>();
@@ -115,6 +116,13 @@ public PageResult<GlobalLockVO> query(GlobalLockParam param) {
         return PageResult.success(list, count, param.getPageNum(), param.getPageSize());
     }
 
+    private String getOrderByCondition() {
+        if("sqlserver".equals(dbType)) {
+            return "gmt_create desc";
+        }
+        return null;
+    }
+
     private String getWhereConditionByParam(GlobalLockParam param, List<Object> sqlParamList) {
         StringBuilder whereConditionBuilder = new StringBuilder();
         if (StringUtils.isNotBlank(param.getXid())) {

server/src/main/java/org/apache/seata/server/console/impl/db/GlobalSessionDBServiceImpl.java

@@ -86,9 +86,10 @@ public PageResult<GlobalSessionVO> query(GlobalSessionParam param) {
 
         List<Object> sqlParamList = new ArrayList<>();
         String whereCondition = getWhereConditionByParam(param, sqlParamList);
+        String orderByCondition = getOrderByCondition();
 
         String sourceSql = LogStoreSqlsFactory.getLogStoreSqls(dbType).getAllGlobalSessionSql(globalTable, whereCondition);
-        String querySessionSql = PageUtil.pageSql(sourceSql, dbType, param.getPageNum(), param.getPageSize());
+        String querySessionSql = PageUtil.pageSql(sourceSql, dbType, param.getPageNum(), param.getPageSize(), orderByCondition);
         String sessionCountSql = PageUtil.countSql(sourceSql, dbType);
 
         List<GlobalSessionVO> list = new ArrayList<>();
@@ -129,6 +130,13 @@ public PageResult<GlobalSessionVO> query(GlobalSessionParam param) {
         return PageResult.success(list, count, param.getPageNum(), param.getPageSize());
     }
 
+    private String getOrderByCondition() {
+        if ("sqlserver".equals(dbType)) {
+            return "gmt_create desc";
+        }
+        return null;
+    }
+
     private String getWhereConditionByParam(GlobalSessionParam param, List<Object> sqlParamList) {
         StringBuilder whereConditionBuilder = new StringBuilder();
         if (StringUtils.isNotBlank(param.getXid())) {