[feature](single replica ingest) support single replica ingest binlog

be/src/common/config.cpp

@@ -1470,6 +1470,10 @@ DEFINE_Int32(workload_policy_check_interval_ms, "500");
 // Ingest binlog work pool size, -1 is disable, 0 is hardware concurrency
 DEFINE_Int32(ingest_binlog_work_pool_size, "-1");
 
+// Ingest binlog distribute work pool size for single-replica fan-out to followers.
+// 0 means auto (hardware concurrency), negative values are invalid and will fail startup.
+DEFINE_Int32(ingest_binlog_distribute_work_pool_size, "0");
+
 // Ingest binlog with persistent connection
 DEFINE_Bool(enable_ingest_binlog_with_persistent_connection, "false");
 

be/src/common/config.h

@@ -1556,6 +1556,10 @@ DECLARE_Bool(enable_flush_file_cache_async);
 // Ingest binlog work pool size
 DECLARE_Int32(ingest_binlog_work_pool_size);
 
+// Ingest binlog distribute work pool size for single-replica fan-out to followers.
+// 0 means auto (hardware concurrency), negative values are invalid and will fail startup.
+DECLARE_Int32(ingest_binlog_distribute_work_pool_size);
+
 // Ingest binlog with persistent connection
 DECLARE_Bool(enable_ingest_binlog_with_persistent_connection);
 

be/src/common/metrics/doris_metrics.cpp

@@ -107,6 +107,13 @@ DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(binlog_compaction_task_running_total, Metri
                                      compaction_task_state_total, Labels({{"type", "binlog"}}));
 DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(binlog_compaction_task_pending_total, MetricUnit::ROWSETS, "",
                                      compaction_task_state_total, Labels({{"type", "binlog"}}));
+DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(binlog_ingest_redundant_rowset_cleanup_success_total,
+                                     MetricUnit::OPERATIONS, "",
+                                     binlog_ingest_redundant_rowset_cleanup_success_total,
+                                     Labels());
+DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(binlog_ingest_redundant_rowset_cleanup_failed_total,
+                                     MetricUnit::OPERATIONS, "",
+                                     binlog_ingest_redundant_rowset_cleanup_failed_total, Labels());
 DEFINE_COUNTER_METRIC_PROTOTYPE_5ARG(cumulative_compaction_task_running_total, MetricUnit::ROWSETS,
                                      "", compaction_task_state_total,
                                      Labels({{"type", "cumulative"}}));
@@ -347,6 +354,10 @@ DorisMetrics::DorisMetrics() : _metric_registry(_s_registry_name) {
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, base_compaction_task_pending_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, binlog_compaction_task_running_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, binlog_compaction_task_pending_total);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity,
+                                binlog_ingest_redundant_rowset_cleanup_success_total);
+    INT_COUNTER_METRIC_REGISTER(_server_metric_entity,
+                                binlog_ingest_redundant_rowset_cleanup_failed_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, cumulative_compaction_task_running_total);
     INT_COUNTER_METRIC_REGISTER(_server_metric_entity, cumulative_compaction_task_pending_total);
 

be/src/common/metrics/doris_metrics.h

@@ -109,6 +109,8 @@ class DorisMetrics {
     IntCounter* base_compaction_task_pending_total = nullptr;
     IntCounter* binlog_compaction_task_running_total = nullptr;
     IntCounter* binlog_compaction_task_pending_total = nullptr;
+    IntCounter* binlog_ingest_redundant_rowset_cleanup_success_total = nullptr;
+    IntCounter* binlog_ingest_redundant_rowset_cleanup_failed_total = nullptr;
     IntCounter* cumulative_compaction_task_running_total = nullptr;
     IntCounter* cumulative_compaction_task_pending_total = nullptr;
 

be/src/service/backend_service.h

@@ -160,6 +160,7 @@ class BaseBackendService : public BackendServiceIf {
     ExecEnv* _exec_env = nullptr;
     std::unique_ptr<AgentServer> _agent_server;
     std::unique_ptr<ThreadPool> _ingest_binlog_workers;
+    std::unique_ptr<ThreadPool> _ingest_binlog_distribute_workers;
 };
 
 // `StorageEngine` mixin for `BaseBackendService`

be/src/service/backend_service_ingest_helper.h

@@ -0,0 +1,76 @@
+// 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.
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string_view>
+#include <unordered_map>
+
+#include "common/status.h"
+#include "util/stopwatch.hpp"
+
+namespace doris {
+
+class StorageEngine;
+class Tablet;
+using TabletSharedPtr = std::shared_ptr<Tablet>;
+class RowsetMeta;
+using RowsetMetaSharedPtr = std::shared_ptr<RowsetMeta>;
+class PendingRowsetGuard;
+
+// Result of committing an ingested rowset. When commit fails, |status| preserves the
+// original error so callers can log detailed diagnostics instead of a generic message.
+struct IngestCommitResult {
+    enum Code {
+        kCommitted,    // Rowset committed successfully.
+        kAlreadyExist, // Same load id already committed a different rowset; do not overwrite.
+        kError,        // Commit failed with a real error.
+    };
+
+    Code code;
+    Status status; // Only meaningful when code == kError.
+
+    IngestCommitResult(Code c);
+    IngestCommitResult(Code c, Status s);
+
+    bool operator==(Code c) const;
+};
+
+// Commit an ingested rowset to the local tablet. Exposed for unit testing of the
+// single-replica ingest binlog retry path.
+IngestCommitResult commit_ingested_rowset(
+        StorageEngine& engine, const TabletSharedPtr& local_tablet, int64_t txn_id,
+        int64_t partition_id, const RowsetMetaSharedPtr& rowset_meta,
+        PendingRowsetGuard pending_rs_guard, MonotonicStopWatch& watch,
+        std::unordered_map<std::string_view, uint64_t>& elapsed_time_map);
+
+// Delete files downloaded during ingest. Exposed for unit testing of the cleanup path.
+Status _delete_downloaded_files(const std::vector<std::string>& files, std::string_view reason,
+                                int64_t txn_id);
+
+class TIngestBinlogRequest;
+class TStatus;
+
+// Ingest a rowset from a peer backend. Exposed for unit testing of the
+// fetch_from_peer validation path.
+void _ingest_binlog_from_peer(StorageEngine& engine, const TIngestBinlogRequest& request,
+                              const TabletSharedPtr& local_tablet, int64_t txn_id,
+                              int64_t partition_id, TStatus& tstatus);
+
+} // namespace doris

be/src/storage/txn/txn_manager.cpp

@@ -176,6 +176,23 @@ Status TxnManager::prepare_txn(TPartitionId partition_id, TTransactionId transac
     // not found load id
     // case 1: user start a new txn, rowset = null
     // case 2: loading txn from meta env
+    // Defensive: if we are about to overwrite an existing entry with a different load id,
+    // something may be wrong with the caller's idempotency. Log it but keep the existing
+    // overwrite behavior to avoid breaking other paths.
+    if (auto key_it = txn_tablet_map.find(key); key_it != txn_tablet_map.end()) {
+        if (auto tablet_it = key_it->second.find(tablet_info); tablet_it != key_it->second.end()) {
+            const auto& old_load_id = tablet_it->second->load_id;
+            if (old_load_id.hi() != load_id.hi() || old_load_id.lo() != load_id.lo()) {
+                LOG(WARNING)
+                        << "prepare_txn overwriting existing txn entry with different load id, "
+                        << "partition_id=" << key.first << ", txn_id=" << key.second
+                        << ", tablet=" << tablet_info.to_string()
+                        << ", old_load_id=" << old_load_id.hi() << ":" << old_load_id.lo()
+                        << ", new_load_id=" << load_id.hi() << ":" << load_id.lo();
+            }
+        }
+    }
+
     auto load_info = std::make_shared<TabletTxnInfo>(load_id, nullptr, ingest);
     load_info->prepare();
     if (!txn_tablet_map.contains(key)) {

be/src/util/debug/leak_annotations.h

@@ -81,8 +81,15 @@ namespace doris::debug {
 
 class ScopedLSANDisabler {
 public:
+#if defined(DORIS_LSAN_ENABLED) && defined(__linux__)
     ScopedLSANDisabler() { __lsan_disable(); }
     ~ScopedLSANDisabler() { __lsan_enable(); }
+#else
+    // User-provided (non-trivial) destructor so the variable is not optimized out
+    // and does not trigger -Wunused-variable when LSAN is disabled.
+    ScopedLSANDisabler() {}
+    ~ScopedLSANDisabler() {}
+#endif
 };
 
 } // namespace doris::debug