fix row size gt 1000 cause error in oracle (apache#4266)

changes/1.5.0.md

@@ -126,6 +126,7 @@ Seata 是一款开源的分布式事务解决方案，提供高性能和简单
   - [[#3912](https://github.com/seata/seata/pull/3912)] 支持通过env配置JVM参数
   - [[#3939](https://github.com/seata/seata/pull/3939)] 使用map优化大量的判断代码
   - [[#3955](https://github.com/seata/seata/pull/3955)] 添加启动banner
+  - [[#3946](https://github.com/seata/seata/pull/3946)] 修改由于修改记录过多导致分支注册及lock释放失败的问题 
   - [[#3949](https://github.com/seata/seata/pull/3949)] `nacos-config.py` 支持默认参数和选择性输入参数
   - [[#3954](https://github.com/seata/seata/pull/3954)] 移除对druid依赖中过期方法的调用
   - [[#3981](https://github.com/seata/seata/pull/3981)] 优化服务端口的优先级设置

changes/en-us/1.5.0.md

@@ -59,6 +59,7 @@
   - [[#3901](https://github.com/seata/seata/pull/3901)] fix seataio/seata-server servlet-api conflict
   - [[#3931](https://github.com/seata/seata/pull/3931)] fix the wrong path and filename when dump the jvm memory for analysis
   - [[#3976](https://github.com/seata/seata/pull/3976)] fix NPE cause by future timeout
+  - [[#3946](https://github.com/seata/seata/pull/3946)] fix register branch and release lock failed when the size of rows that modified is greater than 1000 in oracle 
   - [[#3949](https://github.com/seata/seata/pull/3949)] fix the problem that `nacos-config.py` will not skip blank options. fix bug that split options may cause content loss
   - [[#3988](https://github.com/seata/seata/pull/3988)] fix the problem that nacos not found user when password has special characters
   - [[#3998](https://github.com/seata/seata/pull/3998)] fix the NPE of jedis multi.exec

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

@@ -15,6 +15,9 @@
  */
 package io.seata.core.store.db.sql.lock;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import io.seata.common.exception.NotSupportYetException;
 import io.seata.config.Configuration;
 import io.seata.config.ConfigurationFactory;
@@ -30,6 +33,8 @@
  */
 public class AbstractLockStoreSql implements LockStoreSql {
 
+    private static final int MAX_IN_SIZE = 1000;
+
     /**
      * The constant CONFIG.
      */
@@ -49,6 +54,17 @@ public class AbstractLockStoreSql implements LockStoreSql {
      */
     protected static final String IN_PARAMS_PLACE_HOLD = " #in_params# ";
 
+    /**
+     * The constant LOCK_TABLE_PK_WHERE_CONDITION_PLACE_HOLD.
+     */
+    protected static final String LOCK_TABLE_PK_WHERE_CONDITION_PLACE_HOLD = " #lock_table_pk_where_condition# ";
+
+    /**
+     * The constant LOCK_TABLE_BRANCH_ID_WHERE_CONDITION_PLACE_HOLD.
+     */
+    protected static final String LOCK_TABLE_BRANCH_ID_WHERE_CONDITION_PLACE_HOLD = " #lock_table_branch_id_where_condition# ";
+
+
     /**
      * The constant ALL_COLUMNS.
      * xid, transaction_id, branch_id, resource_id, table_name, pk, row_key, gmt_create, gmt_modified
@@ -70,7 +86,7 @@ public class AbstractLockStoreSql implements LockStoreSql {
      * The constant BATCH_DELETE_LOCK_SQL.
      */
     private static final String BATCH_DELETE_LOCK_SQL = "delete from " + LOCK_TABLE_PLACE_HOLD
-        + " where " + ServerTableColumnsName.LOCK_TABLE_XID + " = ? and " + ServerTableColumnsName.LOCK_TABLE_ROW_KEY + " in (" + IN_PARAMS_PLACE_HOLD + ") ";
+        + " where " + ServerTableColumnsName.LOCK_TABLE_XID + " = ? and (" + LOCK_TABLE_PK_WHERE_CONDITION_PLACE_HOLD + ") ";
 
     /**
      * The constant BATCH_DELETE_LOCK_BY_BRANCH_SQL.
@@ -88,7 +104,7 @@ public class AbstractLockStoreSql implements LockStoreSql {
      * The constant BATCH_DELETE_LOCK_BY_BRANCHS_SQL.
      */
     private static final String BATCH_DELETE_LOCK_BY_BRANCHS_SQL = "delete from " + LOCK_TABLE_PLACE_HOLD
-        + " where " + ServerTableColumnsName.LOCK_TABLE_XID + " = ? and " + ServerTableColumnsName.LOCK_TABLE_BRANCH_ID + " in (" + IN_PARAMS_PLACE_HOLD + ") ";
+        + " where " + ServerTableColumnsName.LOCK_TABLE_XID + " = ? and (" + LOCK_TABLE_BRANCH_ID_WHERE_CONDITION_PLACE_HOLD + ") ";
 
 
     /**
@@ -101,7 +117,7 @@ public class AbstractLockStoreSql implements LockStoreSql {
      * The constant CHECK_LOCK_SQL.
      */
     private static final String CHECK_LOCK_SQL = "select " + ALL_COLUMNS + " from " + LOCK_TABLE_PLACE_HOLD
-        + " where " + ServerTableColumnsName.LOCK_TABLE_ROW_KEY + " in (" + IN_PARAMS_PLACE_HOLD + ")"
+        + " where " + LOCK_TABLE_PK_WHERE_CONDITION_PLACE_HOLD
         + " order by status desc ";
 
     /**
@@ -127,9 +143,11 @@ public String getDeleteLockSql(String lockTable) {
     }
 
     @Override
-    public String getBatchDeleteLockSql(String lockTable, String paramPlaceHold) {
-        return BATCH_DELETE_LOCK_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(IN_PARAMS_PLACE_HOLD,
-            paramPlaceHold);
+    public String getBatchDeleteLockSql(String lockTable, int rowSize) {
+        List<String> pkNameList = new ArrayList<>();
+        pkNameList.add(ServerTableColumnsName.LOCK_TABLE_ROW_KEY);
+        String whereCondition = buildWhereConditionByPKs(pkNameList,rowSize,MAX_IN_SIZE);
+        return BATCH_DELETE_LOCK_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(LOCK_TABLE_PK_WHERE_CONDITION_PLACE_HOLD, whereCondition);
     }
 
     @Override
@@ -138,9 +156,12 @@ public String getBatchDeleteLockSqlByBranch(String lockTable) {
     }
 
     @Override
-    public String getBatchDeleteLockSqlByBranchs(String lockTable, String paramPlaceHold) {
-        return BATCH_DELETE_LOCK_BY_BRANCHS_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(IN_PARAMS_PLACE_HOLD,
-            paramPlaceHold);
+    public String getBatchDeleteLockSqlByBranchs(String lockTable, int branchSize) {
+        List<String> pkNameList = new ArrayList<>();
+        pkNameList.add(ServerTableColumnsName.BRANCH_TABLE_BRANCH_ID);
+        String whereCondition = buildWhereConditionByPKs(pkNameList,branchSize,MAX_IN_SIZE);
+        return BATCH_DELETE_LOCK_BY_BRANCHS_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(LOCK_TABLE_BRANCH_ID_WHERE_CONDITION_PLACE_HOLD,
+            whereCondition);
     }
 
     @Override
@@ -149,13 +170,75 @@ public String getQueryLockSql(String lockTable) {
     }
 
     @Override
-    public String getCheckLockableSql(String lockTable, String paramPlaceHold) {
-        return CHECK_LOCK_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(IN_PARAMS_PLACE_HOLD, paramPlaceHold);
+    public String getCheckLockableSql(String lockTable, int rowSize) {
+        List<String> pkNameList = new ArrayList<>();
+        pkNameList.add(ServerTableColumnsName.LOCK_TABLE_ROW_KEY);
+        String whereCondition = buildWhereConditionByPKs(pkNameList,rowSize,MAX_IN_SIZE);
+        return CHECK_LOCK_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable).replace(LOCK_TABLE_PK_WHERE_CONDITION_PLACE_HOLD, whereCondition);
     }
 
     @Override
     public String getBatchUpdateStatusLockByGlobalSql(String lockTable) {
         return BATCH_UPDATE_STATUS_LOCK_BY_GLOBAL_SQL.replace(LOCK_TABLE_PLACE_HOLD, lockTable);
     }
 
+    /**
+     * each pk is a condition.the result will like :" (id,userCode) in ((?,?),(?,?)) or (id,userCode) in ((?,?),(?,?)
+     * ) or (id,userCode) in ((?,?))"
+     * Build where condition by pks string.
+     *
+     * @param pkNameList pk column name list
+     * @param rowSize    the row size of records
+     * @param maxInSize  the max in size
+     * @return return where condition sql string.the sql can search all related records not just one.
+     */
+    private String buildWhereConditionByPKs(List<String> pkNameList, int rowSize, int maxInSize) {
+        StringBuilder whereStr = new StringBuilder();
+        //we must consider the situation of composite primary key
+        int batchSize = rowSize % maxInSize == 0 ? rowSize / maxInSize : (rowSize / maxInSize) + 1;
+        for (int batch = 0; batch < batchSize; batch++) {
+            if (batch > 0) {
+                whereStr.append(" or ");
+            }
+            if (pkNameList.size() > 1) {
+                whereStr.append("(");
+            }
+            for (int i = 0; i < pkNameList.size(); i++) {
+                if (i > 0) {
+                    whereStr.append(",");
+                }
+                whereStr.append(pkNameList.get(i));
+            }
+            if (pkNameList.size() > 1) {
+                whereStr.append(")");
+            }
+            whereStr.append(" in ( ");
+
+            int eachSize = (batch == batchSize - 1) ? (rowSize % maxInSize == 0 ? maxInSize : rowSize % maxInSize)
+                : maxInSize;
+            for (int i = 0; i < eachSize; i++) {
+                //each row is a bracket
+                if (i > 0) {
+                    whereStr.append(",");
+                }
+                if (pkNameList.size() > 1) {
+                    whereStr.append("(");
+                }
+                for (int x = 0; x < pkNameList.size(); x++) {
+                    if (x > 0) {
+                        whereStr.append(",");
+                    }
+                    whereStr.append("?");
+                }
+                if (pkNameList.size() > 1) {
+                    whereStr.append(")");
+                }
+            }
+            whereStr.append(" )");
+        }
+
+        return whereStr.toString();
+    }
+
+
 }

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

@@ -52,10 +52,10 @@ public interface LockStoreSql {
      * Get batch delete lock sql string.
      *
      * @param lockTable      the lock table
-     * @param paramPlaceHold the param place hold
+     * @param rowSize the size of rowkey
      * @return the string
      */
-    String getBatchDeleteLockSql(String lockTable, String paramPlaceHold);
+    String getBatchDeleteLockSql(String lockTable, int rowSize);
 
     /**
      * Get batch delete lock sql string.
@@ -69,10 +69,10 @@ public interface LockStoreSql {
      * Get batch delete lock sql string.
      *
      * @param lockTable      the lock table
-     * @param paramPlaceHold the param place hold
+     * @param branchSize the size of branch
      * @return the string
      */
-    String getBatchDeleteLockSqlByBranchs(String lockTable, String paramPlaceHold);
+    String getBatchDeleteLockSqlByBranchs(String lockTable, int branchSize);
 
     /**
      * Get query lock sql string.
@@ -86,10 +86,10 @@ public interface LockStoreSql {
      * Get check lock sql string.
      *
      * @param lockTable      the lock table
-     * @param paramPlaceHold the param place hold
+     * @param rowSize the size of rowkey
      * @return the string
      */
-    String getCheckLockableSql(String lockTable, String paramPlaceHold);
+    String getCheckLockableSql(String lockTable, int rowSize);
 
     /**
      * get batch update status lock by global sql