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

…e#46996)


Problem Summary:
Previously, all tables required commit locks during transaction commit,
which helped reduce conflicts at the MetaService level. However, this
approach may not be optimal for all scenarios since only MOW
(Merge-on-Write) tables truly need strict concurrency control.

This PR adds a new config `enable_commit_lock_for_all_tables` (default:
true) to control the commit lock strategy:

- When enabled (default): All tables will acquire commit locks during
transaction commit, which helps reduce conflicts at MetaService level by
queueing transactions at FE level
- When disabled: Only MOW tables will acquire commit locks, which may
improve concurrency for non-MOW tables but could increase conflicts at
MetaService level

The default setting maintains the original behavior to avoid potential
performance impact from increased MetaService conflicts, while providing
flexibility to optimize for different deployment scenarios.

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;
 

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