[Enhancement](txn) Add config flag for transaction edit log outside-lock optimization

Add enable_txn_log_outside_lock config (default true) to control whether
transaction edit log writes happen inside or outside the write lock.

When enabled, edit log entries are enqueued (FIFO) inside the write lock and
awaited outside it, preserving ordering via the batch queue while reducing
lock hold time. This resolves the potential out-of-order edit log issue from
the previous outside-lock optimization.

- Add submitEdit() split API to EditLog for enqueue-only writes
- Make EditLogItem public with await() method
- Add enqueueTransactionState/awaitTransactionState helpers
- Update all 8 persist call sites in DatabaseTransactionMgr

Author: dataroaring

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

@@ -744,6 +744,18 @@ public class Config extends ConfigBase {
             "Txn manager will reject coming txns."})
     public static int max_running_txn_num_per_db = 10000;
 
+    @ConfField(mutable = true, masterOnly = true, description = {
+            "是否将事务的 edit log 写入移到写锁之外以减少锁竞争。"
+                    + "开启后，edit log 条目在写锁内入队（FIFO 保证顺序），"
+                    + "在写锁外等待持久化完成，从而降低写锁持有时间，提高并发事务吞吐量。"
+                    + "默认开启。关闭后使用传统的锁内同步写入模式。",
+            "Whether to move transaction edit log writes outside the write lock to reduce lock contention. "
+                    + "When enabled, edit log entries are enqueued inside the write lock (FIFO preserves ordering) "
+                    + "and awaited outside the lock, reducing write lock hold time "
+                    + "and improving concurrent transaction throughput. "
+                    + "Default is true. Set to false to use the traditional in-lock synchronous write mode."})
+    public static boolean enable_txn_log_outside_lock = true;
+
     @ConfField(masterOnly = true, description = {"pending load task 执行线程数。这个配置可以限制当前等待的导入作业数。"
             + "并且应小于 `max_running_txn_num_per_db`。",
             "The pending load task executor pool size. "

fe/fe-core/src/main/java/org/apache/doris/DorisFE.java

@@ -593,6 +593,11 @@ private static void fuzzyConfigs() {
             LOG.info("fuzzy set random_add_cluster_keys_for_mow={}", Config.random_add_cluster_keys_for_mow);
         }
 
+        Config.enable_txn_log_outside_lock = (LocalDate.now().getDayOfMonth() % 2 == 0);
+        LOG.info("fuzzy set enable_txn_log_outside_lock={}", Config.enable_txn_log_outside_lock);
+        Config.enable_batch_editlog = (LocalDate.now().getDayOfMonth() % 2 == 1);
+        LOG.info("fuzzy set enable_batch_editlog={}", Config.enable_batch_editlog);
+
         setFuzzyForCatalog();
     }
 

fe/fe-core/src/main/java/org/apache/doris/persist/EditLog.java

@@ -124,13 +124,14 @@
 public class EditLog {
     public static final Logger LOG = LogManager.getLogger(EditLog.class);
 
-    // Helper class to hold log edit requests
-    private static class EditLogItem {
+    // Helper class to hold log edit requests.
+    // Public so that callers can enqueue inside a lock and await outside it.
+    public static class EditLogItem {
         static AtomicLong nextUid = new AtomicLong(0);
         final short op;
         final Writable writable;
         final Object lock = new Object();
-        boolean finished = false;
+        volatile boolean finished = false;
         long logId = -1;
         long uid = -1;
 
@@ -139,6 +140,24 @@ private static class EditLogItem {
             this.writable = writable;
             uid = nextUid.getAndIncrement();
         }
+
+        /**
+         * Wait for this edit log entry to be flushed to persistent storage.
+         * Returns the assigned log ID.
+         */
+        public long await() {
+            synchronized (lock) {
+                while (!finished) {
+                    try {
+                        lock.wait();
+                    } catch (InterruptedException e) {
+                        LOG.error("Fatal Error : write stream Exception");
+                        System.exit(-1);
+                    }
+                }
+            }
+            return logId;
+        }
     }
 
     private final BlockingQueue<EditLogItem> logEditQueue = new LinkedBlockingQueue<>();
@@ -1534,6 +1553,49 @@ public long logEditWithQueue(short op, Writable writable) {
         return req.logId;
     }
 
+    /**
+     * Submit an edit log entry to the batch queue without waiting for it to be flushed.
+     * The entry is enqueued in FIFO order, so calling this inside a write lock guarantees
+     * that edit log entries are ordered by lock acquisition order.
+     *
+     * <p>The caller MUST call {@link EditLogItem#await()} after releasing the lock to ensure
+     * the entry is persisted before proceeding.
+     *
+     * <p>If batch edit log is disabled, this falls back to a synchronous direct write
+     * and the returned item is already completed.
+     *
+     * @return an {@link EditLogItem} handle to await completion
+     */
+    public EditLogItem submitEdit(short op, Writable writable) {
+        if (this.getNumEditStreams() == 0) {
+            LOG.error("Fatal Error : no editLog stream", new Exception());
+            throw new Error("Fatal Error : no editLog stream");
+        }
+
+        EditLogItem req = new EditLogItem(op, writable);
+        if (Config.enable_batch_editlog && op != OperationType.OP_TIMESTAMP) {
+            while (true) {
+                try {
+                    logEditQueue.put(req);
+                    break;
+                } catch (InterruptedException e) {
+                    LOG.warn("Interrupted during put, will sleep and retry.");
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException ex) {
+                        LOG.warn("interrupted during sleep, will retry.", ex);
+                    }
+                }
+            }
+        } else {
+            // Non-batch mode: write directly (synchronous)
+            long logId = logEditDirectly(op, writable);
+            req.logId = logId;
+            req.finished = true;
+        }
+        return req;
+    }
+
     private synchronized long logEditDirectly(short op, Writable writable) {
         long logId = -1;
         try {