[opt](load) Add config to control commit lock scope for tables

Author: liaoxin01

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3281,6 +3281,14 @@ public static int metaServiceRpcRetryTimes() {
     @ConfField(mutable = true, description = {"存算分离模式下commit阶段等锁超时时间，默认5s"})
     public static int try_commit_lock_timeout_seconds = 5;
 
+    @ConfField(mutable = true, description = {"是否在事务提交时对所有表启用提交锁。设置为 true 时，所有表都会使用提交锁。"
+            + "设置为 false 时，仅对 Merge-On-Write 表使用提交锁。默认值为 true。",
+            "Whether to enable commit lock for all tables during transaction commit."
+            + "If true, commit lock will be applied to all tables."
+            + "If false, commit lock will only be applied to Merge-On-Write tables."
+            + "Default value is true." })
+    public static boolean enable_commit_lock_for_all_tables = true;
+
     @ConfField(mutable = true, description = {"存算分离模式下是否开启大事务提交，默认false"})
     public static boolean enable_cloud_txn_lazy_commit = false;
 
@@ -3307,11 +3315,11 @@ public static int metaServiceRpcRetryTimes() {
     public static int max_get_tablet_stat_task_threads_num = 4;
 
     @ConfField(mutable = true, description = {"存算分离模式下schema change失败是否重试",
-            "Whether to enable retry when schema change failed in cloud model, default is true."})
+        "Whether to enable retry when schema change failed in cloud model, default is true."})
     public static boolean enable_schema_change_retry_in_cloud_mode = true;
-
+    
     @ConfField(mutable = true, description = {"存算分离模式下schema change重试次数",
-            "Max retry times when schema change failed in cloud model, default is 3."})
+        "Max retry times when schema change failed in cloud model, default is 3."})
     public static int schema_change_max_retry_time = 3;
 
     // ATTN: DONOT add any config not related to cloud mode here

fe/fe-core/src/main/java/org/apache/doris/cloud/transaction/CloudGlobalTransactionMgr.java

@@ -1166,7 +1166,21 @@ private void commitTransactionWithSubTxns(long dbId, List<Table> tableList, long
         executeCommitTxnRequest(commitTxnRequest, transactionId, false, null);
     }
 
-    // add some log and get commit lock, mainly used for mow tables
+    private List<Table> getTablesNeedCommitLock(List<Table> tableList) {
+        if (Config.enable_commit_lock_for_all_tables) {
+            // If enabled, lock all tables
+            return tableList.stream()
+                    .sorted(Comparator.comparingLong(Table::getId))
+                    .collect(Collectors.toList());
+        } else {
+            // If disabled, only lock MOW tables
+            return tableList.stream()
+                    .filter(table -> table instanceof OlapTable && ((OlapTable) table).getEnableUniqueKeyMergeOnWrite())
+                    .sorted(Comparator.comparingLong(Table::getId))
+                    .collect(Collectors.toList());
+        }
+    }
+
     private void beforeCommitTransaction(List<Table> tableList, long transactionId, long timeoutMillis)
             throws UserException {
         for (int i = 0; i < tableList.size(); i++) {
@@ -1180,29 +1194,21 @@ private void beforeCommitTransaction(List<Table> tableList, long transactionId,
             }
         }
 
-        // Get tables that require commit lock - only MOW tables need this:
-        // 1. Filter to keep only OlapTables with MOW enabled
-        // 2. Sort by table ID to maintain consistent locking order and prevent deadlocks
-        List<Table> mowTableList = tableList.stream()
-                .filter(table -> table instanceof OlapTable && ((OlapTable) table).getEnableUniqueKeyMergeOnWrite())
-                .sorted(Comparator.comparingLong(Table::getId))
-                .collect(Collectors.toList());
-        increaseWaitingLockCount(mowTableList);
-        if (!MetaLockUtils.tryCommitLockTables(mowTableList, timeoutMillis, TimeUnit.MILLISECONDS)) {
-            decreaseWaitingLockCount(mowTableList);
+        List<Table> tablesToLock = getTablesNeedCommitLock(tableList);
+        increaseWaitingLockCount(tablesToLock);
+        if (!MetaLockUtils.tryCommitLockTables(tablesToLock, timeoutMillis, TimeUnit.MILLISECONDS)) {
+            decreaseWaitingLockCount(tablesToLock);
             // DELETE_BITMAP_LOCK_ERR will be retried on be
             throw new UserException(InternalErrorCode.DELETE_BITMAP_LOCK_ERR,
                     "get table cloud commit lock timeout, tableList=("
-                            + StringUtils.join(mowTableList, ",") + ")");
+                            + StringUtils.join(tablesToLock, ",") + ")");
         }
     }
 
     private void afterCommitTransaction(List<Table> tableList) {
-        List<Table> mowTableList = tableList.stream()
-                .filter(table -> table instanceof OlapTable && ((OlapTable) table).getEnableUniqueKeyMergeOnWrite())
-                .collect(Collectors.toList());
-        decreaseWaitingLockCount(mowTableList);
-        MetaLockUtils.commitUnlockTables(mowTableList);
+        List<Table> tablesToUnlock = getTablesNeedCommitLock(tableList);
+        decreaseWaitingLockCount(tablesToUnlock);
+        MetaLockUtils.commitUnlockTables(tablesToUnlock);
     }
 
     @Override