From ae06fdef907634d501ec720d46ba86ee6cc2480f Mon Sep 17 00:00:00 2001 From: ryam Date: Wed, 22 Jul 2026 10:42:13 +0800 Subject: [PATCH 1/2] [feature](CCR) support single replica ingest binlog Support single replica ingest binlog for CCR, allowing the leader to download rowset files once and distribute to followers, reducing network traffic and improving ingest performance. Includes SCOPED_ATTACH_TASK compatibility fix for latest master. --- be/src/common/metrics/doris_metrics.cpp | 11 + be/src/common/metrics/doris_metrics.h | 2 + be/src/service/backend_service.cpp | 927 ++++++++++++++++++++---- be/src/service/backend_service.h | 1 + be/src/storage/txn/txn_manager.cpp | 17 + gensrc/thrift/BackendService.thrift | 28 + 6 files changed, 861 insertions(+), 125 deletions(-) diff --git a/be/src/common/metrics/doris_metrics.cpp b/be/src/common/metrics/doris_metrics.cpp index 85317133e12382..fd84ac5216fff6 100644 --- a/be/src/common/metrics/doris_metrics.cpp +++ b/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); diff --git a/be/src/common/metrics/doris_metrics.h b/be/src/common/metrics/doris_metrics.h index 764d5dd956a490..a2682eb849ae72 100644 --- a/be/src/common/metrics/doris_metrics.h +++ b/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; diff --git a/be/src/service/backend_service.cpp b/be/src/service/backend_service.cpp index b4bbb435abdf0f..0ec0e13f3fd461 100644 --- a/be/src/service/backend_service.cpp +++ b/be/src/service/backend_service.cpp @@ -36,6 +36,7 @@ #include #include +#include #include #include #include @@ -49,6 +50,7 @@ #include "cloud/config.h" #include "common/config.h" #include "common/logging.h" +#include "common/metrics/doris_metrics.h" #include "common/status.h" #include "exprs/function/dictionary_factory.h" #include "format/arrow/arrow_row_batch.h" @@ -62,6 +64,8 @@ #include "runtime/fragment_mgr.h" #include "runtime/result_queue_mgr.h" #include "runtime/runtime_profile.h" +#include "service/backend_options.h" +#include "service/backend_service_ingest_helper.h" #include "service/http/http_client.h" #include "storage/olap_common.h" #include "storage/olap_define.h" @@ -75,8 +79,11 @@ #include "storage/tablet/tablet_meta.h" #include "storage/txn/txn_manager.h" #include "udf/python/python_env.h" +#include "util/client_cache.h" +#include "util/debug_points.h" #include "util/defer_op.h" #include "util/threadpool.h" +#include "util/thrift_rpc_helper.h" #include "util/thrift_server.h" #include "util/uid_util.h" #include "util/url_coding.h" @@ -93,6 +100,128 @@ class TTransportException; } // namespace apache namespace doris { + +IngestCommitResult::IngestCommitResult(Code c) : code(c) {} +IngestCommitResult::IngestCommitResult(Code c, Status s) : code(c), status(std::move(s)) {} +bool IngestCommitResult::operator==(Code c) const { + return code == c; +} + +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& elapsed_time_map) { + // Step 7.1: create rowset + RowsetSharedPtr rowset; + auto status = RowsetFactory::create_rowset(local_tablet->tablet_schema(), + local_tablet->tablet_path(), rowset_meta, &rowset); + if (!status) { + LOG(WARNING) << "failed to create rowset from rowset meta for remote tablet" + << ". rowset_id: " << rowset_meta->rowset_id() + << ", rowset_type: " << rowset_meta->rowset_type() + << ", tablet_id=" << rowset_meta->tablet_id() << ", txn_id=" << txn_id + << ", status=" << status.to_string(); + return {IngestCommitResult::kError, std::move(status)}; + } + + // Step 7.2 calculate delete bitmap before commit + auto calc_delete_bitmap_token = engine.calc_delete_bitmap_executor()->create_token(); + DeleteBitmapPtr delete_bitmap = std::make_shared(rowset_meta->tablet_id()); + RowsetIdUnorderedSet pre_rowset_ids; + if (local_tablet->enable_unique_key_merge_on_write()) { + auto beta_rowset = reinterpret_cast(rowset.get()); + std::vector segments; + status = beta_rowset->load_segments(&segments); + if (!status) { + LOG(WARNING) << "failed to load segments from rowset" + << ". rowset_id: " << beta_rowset->rowset_id() << ", txn_id=" << txn_id + << ", status=" << status.to_string(); + return {IngestCommitResult::kError, std::move(status)}; + } + elapsed_time_map.emplace("load_segments", watch.elapsed_time_microseconds()); + if (segments.size() > 1) { + // calculate delete bitmap between segments + status = local_tablet->calc_delete_bitmap_between_segments( + rowset->tablet_schema(), rowset->rowset_id(), segments, delete_bitmap); + if (!status) { + LOG(WARNING) << "failed to calculate delete bitmap" + << ". tablet_id: " << local_tablet->tablet_id() + << ". rowset_id: " << rowset->rowset_id() << ", txn_id=" << txn_id + << ", status=" << status.to_string(); + return {IngestCommitResult::kError, std::move(status)}; + } + elapsed_time_map.emplace("calc_delete_bitmap", watch.elapsed_time_microseconds()); + } + + static_cast(BaseTablet::commit_phase_update_delete_bitmap( + local_tablet, rowset, pre_rowset_ids, delete_bitmap, segments, txn_id, + calc_delete_bitmap_token.get(), nullptr)); + elapsed_time_map.emplace("commit_phase_update_delete_bitmap", + watch.elapsed_time_microseconds()); + static_cast(calc_delete_bitmap_token->wait()); + elapsed_time_map.emplace("wait_delete_bitmap", watch.elapsed_time_microseconds()); + } + + // Step 7.3: commit txn + Status commit_txn_status = engine.txn_manager()->commit_txn( + local_tablet->data_dir()->get_meta(), rowset_meta->partition_id(), + rowset_meta->txn_id(), rowset_meta->tablet_id(), local_tablet->tablet_uid(), + rowset_meta->load_id(), rowset, std::move(pending_rs_guard), false); + elapsed_time_map.emplace("commit_txn", watch.elapsed_time_microseconds()); + + if (!commit_txn_status) { + if (commit_txn_status.is()) { + LOG(INFO) << "find transaction already exist when commit ingested rowset, skip commit." + << " rowset_id: " << rowset_meta->rowset_id().to_string() + << ", tablet_id=" << rowset_meta->tablet_id() + << ", txn_id=" << rowset_meta->txn_id(); + return IngestCommitResult::kAlreadyExist; + } + auto err_msg = fmt::format( + "failed to commit txn for remote tablet. rowset_id: {}, tablet_id={}, " + "txn_id={}, status={}", + rowset_meta->rowset_id().to_string(), rowset_meta->tablet_id(), + rowset_meta->txn_id(), commit_txn_status.to_string()); + LOG(WARNING) << err_msg; + return {IngestCommitResult::kError, std::move(commit_txn_status)}; + } + + if (local_tablet->enable_unique_key_merge_on_write()) { + engine.txn_manager()->set_txn_related_delete_bitmap( + partition_id, txn_id, rowset_meta->tablet_id(), local_tablet->tablet_uid(), true, + delete_bitmap, pre_rowset_ids, nullptr); + elapsed_time_map.emplace("set_txn_related_delete_bitmap", + watch.elapsed_time_microseconds()); + } + + return IngestCommitResult::kCommitted; +} + +// Delete files downloaded during ingest. Returns the deletion status so callers can +// update metrics or decide whether additional action is needed. Does not change the +// caller's transaction result; failures are logged so orphan-file issues remain visible. +Status _delete_downloaded_files(const std::vector& files, std::string_view reason, + int64_t txn_id) { + if (files.empty()) { + return Status::OK(); + } + std::vector paths; + paths.reserve(files.size()); + for (const auto& file : files) { + paths.emplace_back(file); + } + auto st = io::global_local_filesystem()->batch_delete(paths); + if (!st.ok()) { + LOG(WARNING) << "failed to delete " << files.size() << " downloaded files (" << reason + << "), txn_id=" << txn_id << ", status=" << st.to_string(); + } else { + LOG(INFO) << "done delete " << files.size() << " downloaded files (" << reason + << "), txn_id=" << txn_id; + } + return st; +} + namespace { bvar::LatencyRecorder g_ingest_binlog_latency("doris_backend_service", "ingest_binlog"); @@ -104,6 +233,9 @@ struct IngestBinlogArg { TabletSharedPtr local_tablet; TIngestBinlogRequest request; TStatus* tstatus; + std::vector* success_replica_backend_ids = nullptr; + std::vector* failed_replica_backend_ids = nullptr; + ThreadPool* follower_distribute_pool = nullptr; }; Status _exec_http_req(std::optional& client, int retry_times, int sleep_time, @@ -118,7 +250,8 @@ Status _exec_http_req(std::optional& client, int retry_times, int sl Status _download_binlog_segment_file(HttpClient* client, const std::string& get_segment_file_url, const std::string& segment_path, uint64_t segment_file_size, uint64_t estimate_timeout, - std::vector& download_success_files) { + std::vector& download_success_files, + std::string* file_md5 = nullptr) { RETURN_IF_ERROR(client->init(get_segment_file_url)); client->set_timeout_ms(estimate_timeout * 1000); RETURN_IF_ERROR(client->download(segment_path)); @@ -160,6 +293,14 @@ Status _download_binlog_segment_file(HttpClient* client, const std::string& get_ } } + if (file_md5 != nullptr) { + if (remote_file_md5.empty()) { + RETURN_IF_ERROR(io::global_local_filesystem()->md5sum(segment_path, file_md5)); + } else { + *file_md5 = remote_file_md5; + } + } + return io::global_local_filesystem()->permission(segment_path, io::LocalFileSystem::PERMS_OWNER_RW); } @@ -168,7 +309,8 @@ Status _download_binlog_index_file(HttpClient* client, const std::string& get_segment_index_file_url, const std::string& local_segment_index_path, uint64_t segment_index_file_size, uint64_t estimate_timeout, - std::vector& download_success_files) { + std::vector& download_success_files, + std::string* file_md5 = nullptr) { RETURN_IF_ERROR(client->init(get_segment_index_file_url)); client->set_timeout_ms(estimate_timeout * 1000); RETURN_IF_ERROR(client->download(local_segment_index_path)); @@ -212,10 +354,432 @@ Status _download_binlog_index_file(HttpClient* client, } } + if (file_md5 != nullptr) { + if (remote_file_md5.empty()) { + RETURN_IF_ERROR( + io::global_local_filesystem()->md5sum(local_segment_index_path, file_md5)); + } else { + *file_md5 = remote_file_md5; + } + } + return io::global_local_filesystem()->permission(local_segment_index_path, io::LocalFileSystem::PERMS_OWNER_RW); } +Status _download_file_from_peer(const std::string& peer_host, const std::string& peer_http_port, + const std::string& peer_token, const std::string& remote_path, + const std::string& local_path, uint64_t file_size, + const std::string& expected_md5, uint64_t estimate_timeout, + std::vector& download_success_files) { + auto remote_file_url = + fmt::format("http://{}:{}/api/_tablet/_download?token={}&file={}&channel=ingest_binlog", + peer_host, peer_http_port, peer_token, remote_path); + auto download_cb = [&remote_file_url, &local_path, &peer_host, &remote_path, file_size, + estimate_timeout, &expected_md5, + &download_success_files](HttpClient* client) { + RETURN_IF_ERROR(client->init(remote_file_url)); + client->set_timeout_ms(estimate_timeout * 1000); + RETURN_IF_ERROR(client->download(local_path)); + download_success_files.push_back(local_path); + + LOG(INFO) << "download file from peer host=" << peer_host << " path=" << remote_path + << " to " << local_path << ", expected md5: " << expected_md5 + << ", size: " << file_size; + + std::error_code ec; + uint64_t local_file_size = std::filesystem::file_size(local_path, ec); + if (ec) { + LOG(WARNING) << "download file from peer error " << ec.message(); + return Status::IOError("can't retrieve file_size of {}, due to {}", local_path, + ec.message()); + } + if (local_file_size != file_size) { + LOG(WARNING) << "download file from peer length error" + << ", peer_host=" << peer_host << ", remote_path=" << remote_path + << ", file_size=" << file_size << ", local_file_size=" << local_file_size; + return Status::RuntimeError("downloaded file size is not equal, local={}, remote={}", + local_file_size, file_size); + } + + if (!expected_md5.empty()) { + std::string local_file_md5; + RETURN_IF_ERROR(io::global_local_filesystem()->md5sum(local_path, &local_file_md5)); + if (local_file_md5 != expected_md5) { + LOG(WARNING) << "download file from peer md5 error" + << ", peer_host=" << peer_host << ", remote_path=" << remote_path + << ", expected_md5=" << expected_md5 + << ", local_file_md5=" << local_file_md5; + return Status::RuntimeError("downloaded file md5 is not equal, local={}, remote={}", + local_file_md5, expected_md5); + } + } + + return io::global_local_filesystem()->permission(local_path, + io::LocalFileSystem::PERMS_OWNER_RW); + }; + return HttpClient::execute_with_retry(3, 1, download_cb); +} + +void _ingest_binlog_from_peer_impl(StorageEngine& engine, const TIngestBinlogRequest& request, + const TabletSharedPtr& local_tablet, int64_t txn_id, + int64_t partition_id, TStatus& tstatus) { + auto set_tstatus = [&tstatus](TStatusCode::type code, std::string error_msg) { + tstatus.__set_status_code(code); + tstatus.__isset.error_msgs = true; + tstatus.error_msgs.push_back(std::move(error_msg)); + }; + + std::shared_ptr mem_tracker = MemTrackerLimiter::create_shared( + MemTrackerLimiter::Type::OTHER, fmt::format("IngestBinlogFromPeer#TxnId={}", txn_id)); + SCOPED_ATTACH_TASK(mem_tracker); + + auto estimate_download_timeout = [](int64_t file_size) { + uint64_t estimate_timeout = file_size / config::download_low_speed_limit_kbps / 1024; + if (estimate_timeout < config::download_low_speed_time) { + estimate_timeout = config::download_low_speed_time; + } + return estimate_timeout; + }; + + MonotonicStopWatch watch(true); + std::unordered_map elapsed_time_map; + std::vector download_success_files; + bool commit_already_exist = false; + Defer defer {[&engine, &tstatus, txn_id, partition_id, &local_tablet, &download_success_files, + &commit_already_exist]() { + if (tstatus.status_code != TStatusCode::OK) { + engine.txn_manager()->abort_txn(partition_id, txn_id, local_tablet->tablet_id(), + local_tablet->tablet_uid()); + LOG(WARNING) << "will delete downloaded peer files due to error " << tstatus; + static_cast( + _delete_downloaded_files(download_success_files, "peer error cleanup", txn_id)); + return; + } + + // Follower path has no distribution step. If the transaction was already committed, + // the rowset files downloaded in this round are redundant and can be deleted immediately. + if (commit_already_exist && !download_success_files.empty()) { + LOG(INFO) << "will delete redundant peer files for already-committed txn " << txn_id + << ", count=" << download_success_files.size(); + auto cleanup_st = _delete_downloaded_files(download_success_files, + "redundant peer cleanup", txn_id); + if (cleanup_st.ok()) { + DorisMetrics::instance() + ->binlog_ingest_redundant_rowset_cleanup_success_total->increment(1); + } else { + DorisMetrics::instance() + ->binlog_ingest_redundant_rowset_cleanup_failed_total->increment(1); + } + } + }}; + + // Check required fields + if (!request.__isset.rowset_meta || request.rowset_meta.empty()) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "rowset_meta is empty for fetch_from_peer"); + return; + } + if (!request.__isset.files) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "files is not set for fetch_from_peer"); + return; + } + if (!request.__isset.peer_host || request.peer_host.empty()) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "peer_host is empty for fetch_from_peer"); + return; + } + if (!request.__isset.peer_http_port || request.peer_http_port.empty()) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "peer_http_port is empty for fetch_from_peer"); + return; + } + if (!request.__isset.peer_token || request.peer_token.empty()) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "peer_token is empty for fetch_from_peer"); + return; + } + + // Parse rowset meta from leader + RowsetMetaPB rowset_meta_pb; + if (!rowset_meta_pb.ParseFromString(request.rowset_meta)) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "failed to parse rowset_meta from peer"); + return; + } + + // Generate local rowset id and localize tablet uid + RowsetMetaSharedPtr rowset_meta = std::make_shared(); + if (!rowset_meta->init_from_pb(rowset_meta_pb)) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "failed to init rowset meta from peer"); + return; + } + RowsetId new_rowset_id = engine.next_rowset_id(); + auto pending_rs_guard = engine.pending_local_rowsets().add(new_rowset_id); + rowset_meta->set_rowset_id(new_rowset_id); + rowset_meta->set_tablet_uid(local_tablet->tablet_uid()); + rowset_meta->set_tablet_schema_hash(local_tablet->tablet_meta()->schema_hash()); + + // Empty rowset (no segments/data files) is valid: skip download and commit directly. + if (request.files.empty()) { + if (rowset_meta->num_segments() != 0) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, + "files is empty for fetch_from_peer but rowset has segments"); + return; + } + auto commit_result = + commit_ingested_rowset(engine, local_tablet, txn_id, partition_id, rowset_meta, + std::move(pending_rs_guard), watch, elapsed_time_map); + if (commit_result == IngestCommitResult::kError) { + set_tstatus(TStatusCode::RUNTIME_ERROR, + fmt::format("failed to commit empty rowset from peer, status={}", + commit_result.status.to_string())); + return; + } + if (commit_result == IngestCommitResult::kAlreadyExist) { + commit_already_exist = true; + LOG(INFO) << "ingest binlog from peer empty rowset already committed, txn_id=" + << txn_id; + } + tstatus.__set_status_code(TStatusCode::OK); + return; + } + + // Check capacity + uint64_t total_size = 0; + for (const auto& file : request.files) { + if (!file.__isset.size) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, + fmt::format("file size is missing for {}", file.remote_path)); + return; + } + total_size += file.size; + } + if (!local_tablet->can_add_binlog(total_size)) { + set_tstatus(TStatusCode::INTERNAL_ERROR, + fmt::format("failed to add binlog from peer, no enough space, total_size={}", + total_size)); + return; + } + + // Download files from peer + for (const auto& file : request.files) { + if (!file.__isset.remote_path || file.remote_path.empty()) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, "remote_path is empty in peer file info"); + return; + } + if (!file.__isset.segment_index) { + set_tstatus(TStatusCode::ANALYSIS_ERROR, + fmt::format("segment_index is missing for {}", file.remote_path)); + return; + } + + std::string local_path; + if (file.__isset.is_index_file && file.is_index_file) { + auto segment_path = + local_segment_path(local_tablet->tablet_path(), + rowset_meta->rowset_id().to_string(), file.segment_index); + if (file.__isset.index_id && file.index_id != -1) { + // V1 format + std::string suffix_path = file.__isset.suffix_path ? file.suffix_path : ""; + local_path = InvertedIndexDescriptor::get_index_file_path_v1( + InvertedIndexDescriptor::get_index_file_path_prefix(segment_path), + file.index_id, suffix_path); + } else { + // V2 format + local_path = InvertedIndexDescriptor::get_index_file_path_v2( + InvertedIndexDescriptor::get_index_file_path_prefix(segment_path)); + } + } else { + local_path = + local_segment_path(local_tablet->tablet_path(), + rowset_meta->rowset_id().to_string(), file.segment_index); + } + + uint64_t estimate_timeout = estimate_download_timeout(file.size); + std::string expected_md5 = file.__isset.md5 ? file.md5 : ""; + auto status = _download_file_from_peer( + request.peer_host, request.peer_http_port, request.peer_token, file.remote_path, + local_path, file.size, expected_md5, estimate_timeout, download_success_files); + if (!status.ok()) { + set_tstatus(TStatusCode::RUNTIME_ERROR, status.to_string()); + return; + } + } + elapsed_time_map.emplace("download_files_from_peer", watch.elapsed_time_microseconds()); + + // Commit rowset + auto commit_result = + commit_ingested_rowset(engine, local_tablet, txn_id, partition_id, rowset_meta, + std::move(pending_rs_guard), watch, elapsed_time_map); + if (commit_result == IngestCommitResult::kError) { + set_tstatus(TStatusCode::RUNTIME_ERROR, + fmt::format("failed to commit ingested rowset from peer, status={}", + commit_result.status.to_string())); + return; + } + if (commit_result == IngestCommitResult::kAlreadyExist) { + commit_already_exist = true; + LOG(INFO) << "ingest binlog from peer txn already committed, will clean up redundant " + "files, txn_id=" + << txn_id << ", file_count=" << download_success_files.size(); + } + + tstatus.__set_status_code(TStatusCode::OK); +} + +Status _distribute_ingested_rowset_to_followers( + StorageEngine& engine, const TIngestBinlogRequest& request, + const RowsetMetaSharedPtr& rowset_meta, + const std::vector& ingested_files, + std::vector& success_backend_ids, std::vector& failed_backend_ids, + ThreadPool* distribute_pool, const std::shared_ptr& parent_mem_tracker) { + if (!request.__isset.follower_replicas || request.follower_replicas.empty()) { + return Status::OK(); + } + + std::string rowset_meta_str; + if (!rowset_meta->serialize(&rowset_meta_str)) { + return Status::InternalError("failed to serialize rowset meta for followers"); + } + + std::string peer_host = BackendOptions::get_localhost(); + std::string peer_http_port = std::to_string(config::webserver_port); + std::string peer_token = ExecEnv::GetInstance()->token(); + + uint64_t total_file_size = 0; + for (const auto& file : ingested_files) { + if (file.__isset.size) { + total_file_size += file.size; + } + } + uint64_t estimate_timeout_s = total_file_size / config::download_low_speed_limit_kbps / 1024; + if (estimate_timeout_s < config::download_low_speed_time) { + estimate_timeout_s = config::download_low_speed_time; + } + estimate_timeout_s = estimate_timeout_s * 3 / 2; // 1.5x margin + int timeout_ms = static_cast(std::min(estimate_timeout_s, static_cast(7200)) * + 1000); // cap 2h + + // Validate all follower infos before launching any RPC. Invalid ones are recorded as failed + // so that the caller can decide to fallback instead of aborting the already-committed leader. + struct FollowerTask { + int64_t backend_id; + std::string host; + int32_t be_port; + }; + std::vector valid_followers; + valid_followers.reserve(request.follower_replicas.size()); + for (const auto& follower : request.follower_replicas) { + if (!follower.__isset.backend_id || !follower.__isset.host || !follower.__isset.be_port) { + int64_t bad_id = follower.__isset.backend_id ? follower.backend_id : -1; + LOG(WARNING) << "invalid follower replica info, backend_id=" << bad_id; + failed_backend_ids.push_back(bad_id); + continue; + } + valid_followers.push_back({follower.backend_id, follower.host, follower.be_port}); + } + + std::vector>> futures; + futures.reserve(valid_followers.size()); + + for (const auto& task : valid_followers) { + int64_t backend_id = task.backend_id; + std::string host = task.host; + int32_t be_port = task.be_port; + + auto promise_ptr = std::make_shared>>(); + futures.push_back(promise_ptr->get_future()); + + auto worker = [promise_ptr, backend_id, host, be_port, timeout_ms, &request, + &rowset_meta_str, &ingested_files, &peer_host, &peer_http_port, &peer_token, + parent_mem_tracker]() { + SCOPED_ATTACH_TASK(parent_mem_tracker); + try { + TIngestBinlogResult follower_result; + TIngestBinlogRequest follower_request; + follower_request.__set_txn_id(request.txn_id); + follower_request.__set_partition_id(request.partition_id); + follower_request.__set_local_tablet_id(request.local_tablet_id); + follower_request.__set_load_id(request.load_id); + follower_request.__set_fetch_from_peer(true); + follower_request.__set_peer_host(peer_host); + follower_request.__set_peer_http_port(peer_http_port); + follower_request.__set_peer_token(peer_token); + follower_request.__set_rowset_meta(rowset_meta_str); + follower_request.__set_files(ingested_files); + + DBUG_EXECUTE_IF("ingest_binlog.follower.force_fail", { + auto target_backend_id = + DebugPoints::instance()->get_debug_param_or_default( + "ingest_binlog.follower.force_fail", "backend_id", -1); + if (target_backend_id == -1 || target_backend_id == backend_id) { + LOG(WARNING) << "debug point force follower ingest_binlog fail, " + << "backend_id=" << backend_id; + promise_ptr->set_value(std::make_pair( + backend_id, + Status::InternalError("debug point force follower fail"))); + return; + } + }); + + Status status = ThriftRpcHelper::rpc( + host, be_port, + [&follower_request, + &follower_result](ClientConnection& client) { + client->ingest_binlog(follower_result, follower_request); + }, + timeout_ms); + if (!status.ok()) { + LOG(WARNING) << "failed to send ingest_binlog to follower " << host << ":" + << be_port << ", backend_id=" << backend_id + << ", status=" << status.to_string(); + promise_ptr->set_value(std::make_pair(backend_id, status)); + return; + } + if (follower_result.status.status_code != TStatusCode::OK) { + status = Status::create(follower_result.status); + LOG(WARNING) << "follower ingest_binlog failed, backend_id=" << backend_id + << ", status=" << status.to_string(); + promise_ptr->set_value(std::make_pair(backend_id, status)); + return; + } + promise_ptr->set_value(std::make_pair(backend_id, Status::OK())); + } catch (const std::exception& e) { + LOG(WARNING) << "follower ingest_binlog task threw exception, backend_id=" + << backend_id << ", exception=" << e.what(); + promise_ptr->set_value(std::make_pair(backend_id, Status::InternalError(e.what()))); + } + }; + + if (distribute_pool != nullptr) { + Status st = distribute_pool->submit_func(worker); + if (st.ok()) { + continue; + } + // The pool queue is full. Fall back to inline execution in the thrift + // handler thread instead of failing the follower: this transfers + // backpressure to the caller (CCR acquires a per-backend concurrency + // window for every ingest) and avoids spurious whole-txn retries that + // would waste the cross-cluster download this feature saves. + LOG(WARNING) << "ingest binlog follower distribute pool is full, run follower " + "distribution inline, backend_id=" + << backend_id << ", status=" << st.to_string(); + } + worker(); + } + + for (auto& future : futures) { + auto [backend_id, status] = future.get(); + if (status.ok()) { + success_backend_ids.push_back(backend_id); + } else { + failed_backend_ids.push_back(backend_id); + } + } + + if (!failed_backend_ids.empty()) { + return Status::RuntimeError("{} follower(s) failed to ingest from peer", + failed_backend_ids.size()); + } + return Status::OK(); +} + void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { std::optional client; if (config::enable_ingest_binlog_with_persistent_connection) { @@ -241,8 +805,17 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { TStatus tstatus; std::vector download_success_files; std::unordered_map elapsed_time_map; + bool is_single_replica_download = + request.__isset.single_replica_download && request.single_replica_download; + std::vector ingested_files; + bool committed = false; + bool commit_already_exist = false; + bool distribution_done = false; + std::vector redundant_files_to_delete; Defer defer {[=, &engine, &tstatus, ingest_binlog_tstatus = arg->tstatus, &watch, - &total_download_bytes, &total_download_files, &elapsed_time_map]() { + &total_download_bytes, &total_download_files, &elapsed_time_map, + &download_success_files, &committed, &commit_already_exist, &distribution_done, + &redundant_files_to_delete]() { g_ingest_binlog_latency << watch.elapsed_time_microseconds(); auto elapsed_time_ms = watch.elapsed_time_milliseconds(); double copy_rate = 0.0; @@ -263,19 +836,32 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { LOG(WARNING) << "ingest binlog elapsed " << elapsed_time_ms << " ms, " << elapsed_details; } - if (tstatus.status_code != TStatusCode::OK) { + if (tstatus.status_code != TStatusCode::OK && !committed) { // abort txn engine.txn_manager()->abort_txn(partition_id, txn_id, local_tablet_id, local_tablet_uid); // delete all successfully downloaded files LOG(WARNING) << "will delete downloaded success files due to error " << tstatus; - std::vector paths; - for (const auto& file : download_success_files) { - paths.emplace_back(file); - LOG(WARNING) << "will delete downloaded success file " << file << " due to error"; + static_cast(_delete_downloaded_files(download_success_files, + "leader error cleanup", txn_id)); + } + + // When the transaction was already committed by a previous attempt, the rowset files + // downloaded in this round (R2) are redundant after follower distribution completes. + // Delete them to avoid orphan files, but only after distribution is done because + // followers may still be fetching these files via HTTP. + if (commit_already_exist && distribution_done && !redundant_files_to_delete.empty()) { + LOG(INFO) << "will delete redundant rowset files downloaded for already-committed txn " + << txn_id << ", count=" << redundant_files_to_delete.size(); + auto cleanup_st = _delete_downloaded_files(redundant_files_to_delete, + "leader redundant cleanup", txn_id); + if (cleanup_st.ok()) { + DorisMetrics::instance() + ->binlog_ingest_redundant_rowset_cleanup_success_total->increment(1); + } else { + DorisMetrics::instance() + ->binlog_ingest_redundant_rowset_cleanup_failed_total->increment(1); } - static_cast(io::global_local_filesystem()->batch_delete(paths)); - LOG(WARNING) << "done delete downloaded success files due to error " << tstatus; } if (ingest_binlog_tstatus) { @@ -283,12 +869,6 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { } }}; - auto set_tstatus = [&tstatus](TStatusCode::type code, std::string error_msg) { - tstatus.__set_status_code(code); - tstatus.__isset.error_msgs = true; - tstatus.error_msgs.push_back(std::move(error_msg)); - }; - auto estimate_download_timeout = [](int64_t file_size) { uint64_t estimate_timeout = file_size / config::download_low_speed_limit_kbps / 1024; if (estimate_timeout < config::download_low_speed_time) { @@ -376,6 +956,11 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { rowset_meta_pb.set_tablet_schema_hash(local_tablet->tablet_meta()->schema_hash()); rowset_meta_pb.set_txn_id(txn_id); rowset_meta_pb.set_rowset_state(RowsetStatePB::COMMITTED); + // Unify load id: both prepare_txn and commit_txn use the load id from the ingest request, + // so retries of the same transaction hit the idempotent short-circuit instead of replacing + // the already-committed rowset. + rowset_meta_pb.mutable_load_id()->set_hi(request.load_id.hi); + rowset_meta_pb.mutable_load_id()->set_lo(request.load_id.lo); auto rowset_meta = std::make_shared(); if (!rowset_meta->init_from_pb(rowset_meta_pb)) { LOG(WARNING) << "failed to init rowset meta from " << get_rowset_meta_url; @@ -446,11 +1031,15 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { LOG(INFO) << "download segment file from " << get_segment_file_url << " to " << segment_path; uint64_t estimate_timeout = estimate_download_timeout(segment_file_size); + std::string segment_file_md5; + std::string* segment_file_md5_ptr = + is_single_replica_download ? &segment_file_md5 : nullptr; auto get_segment_file_cb = [&get_segment_file_url, &segment_path, segment_file_size, - estimate_timeout, &download_success_files](HttpClient* client) { + estimate_timeout, &download_success_files, + segment_file_md5_ptr](HttpClient* client) { return _download_binlog_segment_file(client, get_segment_file_url, segment_path, segment_file_size, estimate_timeout, - download_success_files); + download_success_files, segment_file_md5_ptr); }; status = _exec_http_req(client, max_retry, 1, get_segment_file_cb); @@ -460,6 +1049,19 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { status.to_thrift(&tstatus); return; } + + if (is_single_replica_download) { + TIngestedFileInfo file_info; + file_info.__set_remote_path(segment_path); + file_info.__set_size(segment_file_size); + file_info.__set_segment_index(static_cast(segment_index)); + file_info.__set_index_id(-1); + file_info.__set_is_index_file(false); + if (!segment_file_md5.empty()) { + file_info.__set_md5(segment_file_md5); + } + ingested_files.push_back(std::move(file_info)); + } } elapsed_time_map.emplace("get_segment_files", watch.elapsed_time_microseconds()); @@ -468,6 +1070,9 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { std::vector segment_index_file_urls; std::vector segment_index_file_sizes; std::vector segment_index_file_names; + std::vector segment_index_file_segment_indices; + std::vector segment_index_file_index_ids; + std::vector segment_index_file_suffix_paths; auto tablet_schema = rowset_meta->tablet_schema(); if (tablet_schema->get_inverted_index_storage_format() == InvertedIndexStorageFormatPB::V1) { for (const auto& index : tablet_schema->inverted_indexes()) { @@ -494,6 +1099,9 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { segment_index_file_names.push_back(InvertedIndexDescriptor::get_index_file_path_v1( InvertedIndexDescriptor::get_index_file_path_prefix(segment_path), index_id, index->get_index_suffix())); + segment_index_file_segment_indices.push_back(static_cast(segment_index)); + segment_index_file_index_ids.push_back(index_id); + segment_index_file_suffix_paths.push_back(index->get_index_suffix()); status = _exec_http_req(client, max_retry, 1, get_segment_index_file_size_cb); if (!status.ok()) { @@ -530,6 +1138,9 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { rowset_meta->rowset_id().to_string(), segment_index); segment_index_file_names.push_back(InvertedIndexDescriptor::get_index_file_path_v2( InvertedIndexDescriptor::get_index_file_path_prefix(segment_path))); + segment_index_file_segment_indices.push_back(static_cast(segment_index)); + segment_index_file_index_ids.push_back(-1); + segment_index_file_suffix_paths.emplace_back(); status = _exec_http_req(client, max_retry, 1, get_segment_index_file_size_cb); if (!status.ok()) { @@ -564,6 +1175,9 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { // Step 6.3: get all segment index files DCHECK(segment_index_file_sizes.size() == segment_index_file_names.size()); DCHECK(segment_index_file_names.size() == segment_index_file_urls.size()); + DCHECK(segment_index_file_names.size() == segment_index_file_segment_indices.size()); + DCHECK(segment_index_file_names.size() == segment_index_file_index_ids.size()); + DCHECK(segment_index_file_names.size() == segment_index_file_suffix_paths.size()); for (int64_t i = 0; i < segment_index_file_urls.size(); ++i) { auto segment_index_file_size = segment_index_file_sizes[i]; auto get_segment_index_file_url = segment_index_file_urls[i]; @@ -576,12 +1190,16 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { auto local_segment_index_path = segment_index_file_names[i]; LOG(INFO) << fmt::format("download segment index file from {} to {}", get_segment_index_file_url, local_segment_index_path); + std::string index_file_md5; + std::string* index_file_md5_ptr = is_single_replica_download ? &index_file_md5 : nullptr; auto get_segment_index_file_cb = [&get_segment_index_file_url, &local_segment_index_path, segment_index_file_size, estimate_timeout, - &download_success_files](HttpClient* client) { + &download_success_files, + index_file_md5_ptr](HttpClient* client) { return _download_binlog_index_file(client, get_segment_index_file_url, local_segment_index_path, segment_index_file_size, - estimate_timeout, download_success_files); + estimate_timeout, download_success_files, + index_file_md5_ptr); }; status = _exec_http_req(client, max_retry, 1, get_segment_index_file_cb); @@ -591,88 +1209,65 @@ void _ingest_binlog(StorageEngine& engine, IngestBinlogArg* arg) { status.to_thrift(&tstatus); return; } + + if (is_single_replica_download) { + TIngestedFileInfo file_info; + file_info.__set_remote_path(local_segment_index_path); + file_info.__set_size(segment_index_file_size); + file_info.__set_segment_index(segment_index_file_segment_indices[i]); + file_info.__set_index_id(segment_index_file_index_ids[i]); + file_info.__set_suffix_path(segment_index_file_suffix_paths[i]); + file_info.__set_is_index_file(true); + if (!index_file_md5.empty()) { + file_info.__set_md5(index_file_md5); + } + ingested_files.push_back(std::move(file_info)); + } } elapsed_time_map.emplace("get_segment_index_files", watch.elapsed_time_microseconds()); // Step 7: create rowset && calculate delete bitmap && commit - // Step 7.1: create rowset - RowsetSharedPtr rowset; - status = RowsetFactory::create_rowset(local_tablet->tablet_schema(), - local_tablet->tablet_path(), rowset_meta, &rowset); - if (!status) { - LOG(WARNING) << "failed to create rowset from rowset meta for remote tablet" - << ". rowset_id: " << rowset_meta_pb.rowset_id() - << ", rowset_type: " << rowset_meta_pb.rowset_type() - << ", remote_tablet_id=" << rowset_meta_pb.tablet_id() << ", txn_id=" << txn_id - << ", status=" << status.to_string(); + auto commit_result = + commit_ingested_rowset(engine, local_tablet, txn_id, partition_id, rowset_meta, + std::move(pending_rs_guard), watch, elapsed_time_map); + if (commit_result == IngestCommitResult::kError) { + status = Status::RuntimeError("failed to commit ingested rowset, status={}", + commit_result.status.to_string()); status.to_thrift(&tstatus); return; } + if (commit_result == IngestCommitResult::kAlreadyExist) { + commit_already_exist = true; + // The current round files (R2) are redundant on the leader because a previous attempt + // already committed R1. We still need them for follower distribution below; schedule + // cleanup after distribution completes. + redundant_files_to_delete.assign(download_success_files.begin(), + download_success_files.end()); + LOG(INFO) << "ingest binlog txn already committed, will distribute current files to " + "followers and then clean up redundant files, txn_id=" + << txn_id << ", file_count=" << redundant_files_to_delete.size(); + } else { + committed = true; + } - // Step 7.2 calculate delete bitmap before commit - auto calc_delete_bitmap_token = engine.calc_delete_bitmap_executor()->create_token(); - DeleteBitmapPtr delete_bitmap = std::make_shared(local_tablet_id); - RowsetIdUnorderedSet pre_rowset_ids; - if (local_tablet->enable_unique_key_merge_on_write()) { - auto beta_rowset = reinterpret_cast(rowset.get()); - std::vector segments; - status = beta_rowset->load_segments(&segments); + // Step 8: distribute to followers if single replica download + if (is_single_replica_download) { + DCHECK(arg->success_replica_backend_ids != nullptr); + DCHECK(arg->failed_replica_backend_ids != nullptr); + status = _distribute_ingested_rowset_to_followers( + engine, request, rowset_meta, ingested_files, *arg->success_replica_backend_ids, + *arg->failed_replica_backend_ids, arg->follower_distribute_pool, mem_tracker); if (!status) { - LOG(WARNING) << "failed to load segments from rowset" - << ". rowset_id: " << beta_rowset->rowset_id() << ", txn_id=" << txn_id + LOG(WARNING) << "distribute ingested rowset to followers partially failed, success=" + << arg->success_replica_backend_ids->size() + << ", failed=" << arg->failed_replica_backend_ids->size() << ", status=" << status.to_string(); - status.to_thrift(&tstatus); - return; - } - elapsed_time_map.emplace("load_segments", watch.elapsed_time_microseconds()); - if (segments.size() > 1) { - // calculate delete bitmap between segments - status = local_tablet->calc_delete_bitmap_between_segments( - rowset->tablet_schema(), rowset->rowset_id(), segments, delete_bitmap); - if (!status) { - LOG(WARNING) << "failed to calculate delete bitmap" - << ". tablet_id: " << local_tablet->tablet_id() - << ". rowset_id: " << rowset->rowset_id() << ", txn_id=" << txn_id - << ", status=" << status.to_string(); - status.to_thrift(&tstatus); - return; - } - elapsed_time_map.emplace("calc_delete_bitmap", watch.elapsed_time_microseconds()); + // Do NOT set tstatus to error and do NOT delete files. The rowset is already + // committed on the leader; downstream syncer will retry/fallback based on the + // success/failed replica backend id lists. } - - static_cast(BaseTablet::commit_phase_update_delete_bitmap( - local_tablet, rowset, pre_rowset_ids, delete_bitmap, segments, txn_id, - calc_delete_bitmap_token.get(), nullptr)); - elapsed_time_map.emplace("commit_phase_update_delete_bitmap", - watch.elapsed_time_microseconds()); - static_cast(calc_delete_bitmap_token->wait()); - elapsed_time_map.emplace("wait_delete_bitmap", watch.elapsed_time_microseconds()); - } - - // Step 7.3: commit txn - Status commit_txn_status = engine.txn_manager()->commit_txn( - local_tablet->data_dir()->get_meta(), rowset_meta->partition_id(), - rowset_meta->txn_id(), rowset_meta->tablet_id(), local_tablet->tablet_uid(), - rowset_meta->load_id(), rowset, std::move(pending_rs_guard), false); - if (!commit_txn_status && !commit_txn_status.is()) { - auto err_msg = fmt::format( - "failed to commit txn for remote tablet. rowset_id: {}, remote_tablet_id={}, " - "txn_id={}, status={}", - rowset_meta->rowset_id().to_string(), rowset_meta->tablet_id(), - rowset_meta->txn_id(), commit_txn_status.to_string()); - LOG(WARNING) << err_msg; - set_tstatus(TStatusCode::RUNTIME_ERROR, std::move(err_msg)); - return; - } - elapsed_time_map.emplace("commit_txn", watch.elapsed_time_microseconds()); - - if (local_tablet->enable_unique_key_merge_on_write()) { - engine.txn_manager()->set_txn_related_delete_bitmap(partition_id, txn_id, local_tablet_id, - local_tablet->tablet_uid(), true, - delete_bitmap, pre_rowset_ids, nullptr); - elapsed_time_map.emplace("set_txn_related_delete_bitmap", - watch.elapsed_time_microseconds()); } + distribution_done = true; tstatus.__set_status_code(TStatusCode::OK); } @@ -693,19 +1288,41 @@ Status BackendService::start_thrift_dependencies() { auto thread_num = config::ingest_binlog_work_pool_size; if (thread_num < 0) { - LOG(INFO) << fmt::format("ingest binlog thread pool size is {}, so we will in sync mode", + LOG(INFO) << fmt::format("ingest binlog work pool size is {}, so we will in sync mode", thread_num); - return Status::OK(); + } else { + if (thread_num == 0) { + thread_num = std::thread::hardware_concurrency(); + } + RETURN_IF_ERROR(doris::ThreadPoolBuilder("IngestBinlog") + .set_min_threads(thread_num) + .set_max_threads(thread_num * 2) + .build(&_ingest_binlog_workers)); + LOG(INFO) << fmt::format("ingest binlog work pool size is {}, in async mode", thread_num); } - if (thread_num == 0) { - thread_num = std::thread::hardware_concurrency(); + // Always create the follower distribution pool for single-replica ingest binlog, + // regardless of whether the legacy async ingest pool is enabled. This turns follower + // fan-out from serial RPC execution into parallel execution bounded by the pool size. + // When the pool queue is full, the follower task falls back to inline execution + // instead of being rejected, so a busy pool never fails an ingest by itself. + auto distribute_thread_num = config::ingest_binlog_distribute_work_pool_size; + if (distribute_thread_num < 0) { + return Status::InvalidArgument( + "ingest_binlog_distribute_work_pool_size must be non-negative, got {}", + distribute_thread_num); } - static_cast(doris::ThreadPoolBuilder("IngestBinlog") - .set_min_threads(thread_num) - .set_max_threads(thread_num * 2) - .build(&_ingest_binlog_workers)); - LOG(INFO) << fmt::format("ingest binlog thread pool size is {}, in async mode", thread_num); + if (distribute_thread_num == 0) { + auto hc = static_cast(std::thread::hardware_concurrency()); + distribute_thread_num = hc > 0 ? hc : 1; + } + RETURN_IF_ERROR(doris::ThreadPoolBuilder("IngestBinlogDistribute") + .set_min_threads(0) + .set_max_threads(distribute_thread_num) + .set_max_queue_size(distribute_thread_num * 4) + .build(&_ingest_binlog_distribute_workers)); + LOG(INFO) << fmt::format("ingest binlog distribute work pool size is {}", + distribute_thread_num); return Status::OK(); } @@ -936,7 +1553,13 @@ void BackendService::release_snapshot(TAgentResult& return_value, void BackendService::ingest_binlog(TIngestBinlogResult& result, const TIngestBinlogRequest& request) { - LOG(INFO) << "ingest binlog. request: " << apache::thrift::ThriftDebugString(request); + LOG(INFO) << "ingest binlog. txn_id=" << (request.__isset.txn_id ? request.txn_id : -1) + << ", tablet_id=" << (request.__isset.local_tablet_id ? request.local_tablet_id : -1) + << ", load_id=" << (request.__isset.load_id ? print_id(request.load_id) : "not_set") + << ", fetch_from_peer=" + << (request.__isset.fetch_from_peer && request.fetch_from_peer) + << ", single_replica_download=" + << (request.__isset.single_replica_download && request.single_replica_download); TStatus tstatus; Defer defer {[&result, &tstatus]() { @@ -955,37 +1578,15 @@ void BackendService::ingest_binlog(TIngestBinlogResult& result, return; } - /// Check args: txn_id, remote_tablet_id, binlog_version, remote_host, remote_port, partition_id, load_id + bool is_fetch_from_peer = request.__isset.fetch_from_peer && request.fetch_from_peer; + + /// Check common args: txn_id, partition_id, local_tablet_id, load_id if (!request.__isset.txn_id) { auto error_msg = "txn_id is empty"; LOG(WARNING) << error_msg; set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); return; } - if (!request.__isset.remote_tablet_id) { - auto error_msg = "remote_tablet_id is empty"; - LOG(WARNING) << error_msg; - set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); - return; - } - if (!request.__isset.binlog_version) { - auto error_msg = "binlog_version is empty"; - LOG(WARNING) << error_msg; - set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); - return; - } - if (!request.__isset.remote_host) { - auto error_msg = "remote_host is empty"; - LOG(WARNING) << error_msg; - set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); - return; - } - if (!request.__isset.remote_port) { - auto error_msg = "remote_port is empty"; - LOG(WARNING) << error_msg; - set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); - return; - } if (!request.__isset.partition_id) { auto error_msg = "partition_id is empty"; LOG(WARNING) << error_msg; @@ -1005,6 +1606,34 @@ void BackendService::ingest_binlog(TIngestBinlogResult& result, return; } + // For leader/old path, remote info is required + if (!is_fetch_from_peer) { + if (!request.__isset.remote_tablet_id) { + auto error_msg = "remote_tablet_id is empty"; + LOG(WARNING) << error_msg; + set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); + return; + } + if (!request.__isset.binlog_version) { + auto error_msg = "binlog_version is empty"; + LOG(WARNING) << error_msg; + set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); + return; + } + if (!request.__isset.remote_host) { + auto error_msg = "remote_host is empty"; + LOG(WARNING) << error_msg; + set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); + return; + } + if (!request.__isset.remote_port) { + auto error_msg = "remote_port is empty"; + LOG(WARNING) << error_msg; + set_tstatus(TStatusCode::ANALYSIS_ERROR, error_msg); + return; + } + } + auto txn_id = request.txn_id; // Step 1: get local tablet auto const& local_tablet_id = request.local_tablet_id; @@ -1035,7 +1664,47 @@ void BackendService::ingest_binlog(TIngestBinlogResult& result, } } + // Dispatch by mode + if (is_fetch_from_peer) { + // Follower mode: always synchronous + _ingest_binlog_from_peer_impl(_engine, request, local_tablet, txn_id, partition_id, + tstatus); + return; + } + + bool is_single_replica_download = + request.__isset.single_replica_download && request.single_replica_download; bool is_async = (_ingest_binlog_workers != nullptr); + + if (is_single_replica_download) { + if (is_async) { + set_tstatus(TStatusCode::RUNTIME_ERROR, + "single_replica_download is not supported in async ingest mode"); + return; + } + // Leader mode: synchronous, collect follower results + std::vector success_backend_ids; + std::vector failed_backend_ids; + IngestBinlogArg ingest_binlog_arg = { + .txn_id = txn_id, + .partition_id = partition_id, + .local_tablet_id = local_tablet_id, + .local_tablet = local_tablet, + .request = request, + .tstatus = &tstatus, + .success_replica_backend_ids = &success_backend_ids, + .failed_replica_backend_ids = &failed_backend_ids, + .follower_distribute_pool = _ingest_binlog_distribute_workers.get(), + }; + _ingest_binlog(_engine, &ingest_binlog_arg); + // Always return the per-replica result lists so that syncer can decide whether + // to retry / fallback, even when some followers failed after leader commit. + result.__set_success_replica_backend_ids(success_backend_ids); + result.__set_failed_replica_backend_ids(failed_backend_ids); + return; + } + + // Old path result.__set_is_async(is_async); auto ingest_binlog_func = [=, this, tstatus = &tstatus]() { @@ -1314,4 +1983,12 @@ void BaseBackendService::get_python_packages(std::vector& re result = manager.package_infos_to_thrift(packages); } +// Exposed wrapper for unit tests. The implementation lives in the unnamed namespace +// and is not directly visible to other translation units. +void _ingest_binlog_from_peer(StorageEngine& engine, const TIngestBinlogRequest& request, + const TabletSharedPtr& local_tablet, int64_t txn_id, + int64_t partition_id, TStatus& tstatus) { + _ingest_binlog_from_peer_impl(engine, request, local_tablet, txn_id, partition_id, tstatus); +} + } // namespace doris diff --git a/be/src/service/backend_service.h b/be/src/service/backend_service.h index 5f9c01f5ec0170..9d999d66627afb 100644 --- a/be/src/service/backend_service.h +++ b/be/src/service/backend_service.h @@ -160,6 +160,7 @@ class BaseBackendService : public BackendServiceIf { ExecEnv* _exec_env = nullptr; std::unique_ptr _agent_server; std::unique_ptr _ingest_binlog_workers; + std::unique_ptr _ingest_binlog_distribute_workers; }; // `StorageEngine` mixin for `BaseBackendService` diff --git a/be/src/storage/txn/txn_manager.cpp b/be/src/storage/txn/txn_manager.cpp index 21f8ebb4a4e013..85548918a9b5f2 100644 --- a/be/src/storage/txn/txn_manager.cpp +++ b/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(load_id, nullptr, ingest); load_info->prepare(); if (!txn_tablet_map.contains(key)) { diff --git a/gensrc/thrift/BackendService.thrift b/gensrc/thrift/BackendService.thrift index 3867cec245f12d..bcc4747a027beb 100644 --- a/gensrc/thrift/BackendService.thrift +++ b/gensrc/thrift/BackendService.thrift @@ -234,6 +234,22 @@ struct TWarmUpTabletsResponse { 5: optional i64 finish_job_size } +struct TReplicaDistributionInfo { + 1: optional i64 backend_id; + 2: optional string host; + 3: optional i32 be_port; +} + +struct TIngestedFileInfo { + 1: optional string remote_path; + 2: optional i64 size; + 3: optional i32 segment_index; + 4: optional i64 index_id; + 5: optional string suffix_path; + 6: optional bool is_index_file; + 7: optional string md5; +} + struct TIngestBinlogRequest { 1: optional i64 txn_id; 2: optional i64 remote_tablet_id; @@ -243,11 +259,23 @@ struct TIngestBinlogRequest { 6: optional i64 partition_id; 7: optional i64 local_tablet_id; 8: optional Types.TUniqueId load_id; + // ---- single replica ingest binlog ---- + 9: optional bool single_replica_download; + 10: optional list follower_replicas; + 11: optional bool fetch_from_peer; + 12: optional string peer_host; + 13: optional string peer_http_port; + 14: optional string peer_token; + 15: optional binary rowset_meta; + 16: optional list files; } struct TIngestBinlogResult { 1: optional Status.TStatus status; 2: optional bool is_async; + // ---- single replica ingest binlog ---- + 3: optional list success_replica_backend_ids; + 4: optional list failed_replica_backend_ids; } struct TQueryIngestBinlogRequest { From 61d9ad572e8b2f8a7f92627f4a1c8b8801fea192 Mon Sep 17 00:00:00 2001 From: ryam Date: Wed, 26 Aug 2026 17:38:47 +0800 Subject: [PATCH 2/2] [test](single replica ingest) add BE unit test and regression cases --- be/src/common/config.cpp | 4 + be/src/common/config.h | 4 + .../service/backend_service_ingest_helper.h | 76 ++++ be/src/util/debug/leak_annotations.h | 7 + .../service/backend_service_ingest_test.cpp | 378 ++++++++++++++++ .../doris/regression/suite/Syncer.groovy | 164 ++++++- .../suite/client/BackendClientImpl.groovy | 4 +- .../test_single_replica_ingest_binlog.groovy | 406 ++++++++++++++++++ 8 files changed, 1041 insertions(+), 2 deletions(-) create mode 100644 be/src/service/backend_service_ingest_helper.h create mode 100644 be/test/service/backend_service_ingest_test.cpp create mode 100644 regression-test/suites/ccr_syncer_p0/test_single_replica_ingest_binlog.groovy diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp index ca20cb63990447..87ee78bd51a455 100644 --- a/be/src/common/config.cpp +++ b/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"); diff --git a/be/src/common/config.h b/be/src/common/config.h index 2a6b21ccf8c0e3..59f32225a9e858 100644 --- a/be/src/common/config.h +++ b/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); diff --git a/be/src/service/backend_service_ingest_helper.h b/be/src/service/backend_service_ingest_helper.h new file mode 100644 index 00000000000000..8aa39267717fb4 --- /dev/null +++ b/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 +#include +#include +#include + +#include "common/status.h" +#include "util/stopwatch.hpp" + +namespace doris { + +class StorageEngine; +class Tablet; +using TabletSharedPtr = std::shared_ptr; +class RowsetMeta; +using RowsetMetaSharedPtr = std::shared_ptr; +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& elapsed_time_map); + +// Delete files downloaded during ingest. Exposed for unit testing of the cleanup path. +Status _delete_downloaded_files(const std::vector& 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 diff --git a/be/src/util/debug/leak_annotations.h b/be/src/util/debug/leak_annotations.h index 16a289d85a011c..6875b30b759983 100644 --- a/be/src/util/debug/leak_annotations.h +++ b/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 diff --git a/be/test/service/backend_service_ingest_test.cpp b/be/test/service/backend_service_ingest_test.cpp new file mode 100644 index 00000000000000..a4210dd1972135 --- /dev/null +++ b/be/test/service/backend_service_ingest_test.cpp @@ -0,0 +1,378 @@ +// 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. + +#include +#include +#include + +#include +#include +#include +#include + +#include "common/config.h" +#include "core/block/block.h" +#include "service/backend_service_ingest_helper.h" +#include "storage/data_dir.h" +#include "storage/olap_meta.h" +#include "storage/options.h" +#include "storage/rowset/rowset.h" +#include "storage/rowset/rowset_factory.h" +#include "storage/rowset/rowset_meta.h" +#include "storage/rowset/rowset_meta_manager.h" +#include "storage/rowset/rowset_writer.h" +#include "storage/rowset/rowset_writer_context.h" +#include "storage/storage_engine.h" +#include "storage/tablet/tablet.h" +#include "storage/tablet/tablet_manager.h" +#include "storage/tablet/tablet_meta.h" +#include "storage/tablet/tablet_schema.h" +#include "storage/txn/txn_manager.h" +#include "util/uid_util.h" + +namespace doris { + +static const std::string kTestDir = "./be/test/service/backend_service_ingest_test_data"; +static const int64_t kTabletId = 40001; +static const int64_t kPartitionId = 30001; +static const int64_t kTxnId = 50001; +static const int64_t kSchemaHash = 1111; + +class BackendServiceIngestTest : public testing::Test { +public: + void SetUp() override { + config::txn_map_shard_size = 1; + config::txn_shard_size = 1; + + char buffer[1024]; + EXPECT_NE(getcwd(buffer, 1024), nullptr); + config::storage_root_path = std::string(buffer) + "/backend_service_ingest_test_meta"; + + std::filesystem::remove_all(config::storage_root_path); + std::filesystem::remove_all(kTestDir); + EXPECT_TRUE(std::filesystem::create_directory(config::storage_root_path)); + EXPECT_TRUE(std::filesystem::create_directory(kTestDir)); + + std::vector paths; + paths.emplace_back(config::storage_root_path, -1); + EngineOptions options; + options.store_paths = paths; + options.backend_uid = UniqueId::gen_uid(); + + auto engine = std::make_unique(options); + Status st = engine->open(); + ASSERT_TRUE(st.ok()) << st.to_string(); + _engine = engine.get(); + ExecEnv::GetInstance()->set_storage_engine(std::move(engine)); + + _data_dir = std::make_unique(*_engine, kTestDir); + st = _data_dir->init(); + ASSERT_TRUE(st.ok()) << st.to_string(); + static_cast(_data_dir->update_capacity()); + + _create_mow_tablet(); + } + + void TearDown() override { + ExecEnv::GetInstance()->set_storage_engine(nullptr); + _engine = nullptr; + _data_dir.reset(); + std::filesystem::remove_all(config::storage_root_path); + std::filesystem::remove_all(kTestDir); + } + +protected: + void _create_mow_tablet_schema(TabletSchemaSPtr tablet_schema) { + TabletSchemaPB tablet_schema_pb; + tablet_schema_pb.set_keys_type(UNIQUE_KEYS); + tablet_schema_pb.set_num_short_key_columns(1); + tablet_schema_pb.set_num_rows_per_row_block(1024); + tablet_schema_pb.set_compress_kind(COMPRESS_NONE); + tablet_schema_pb.set_next_column_unique_id(3); + tablet_schema_pb.set_sequence_col_idx(2); + + ColumnPB* column_1 = tablet_schema_pb.add_column(); + column_1->set_unique_id(1); + column_1->set_name("k1"); + column_1->set_type("INT"); + column_1->set_is_key(true); + column_1->set_length(4); + column_1->set_index_length(4); + column_1->set_is_nullable(false); + + ColumnPB* column_2 = tablet_schema_pb.add_column(); + column_2->set_unique_id(2); + column_2->set_name("v1"); + column_2->set_type("INT"); + column_2->set_is_key(false); + column_2->set_length(4); + column_2->set_is_nullable(false); + column_2->set_aggregation("REPLACE"); + + ColumnPB* column_3 = tablet_schema_pb.add_column(); + column_3->set_unique_id(3); + column_3->set_name(SEQUENCE_COL); + column_3->set_type("INT"); + column_3->set_is_key(false); + column_3->set_length(4); + column_3->set_is_nullable(false); + column_3->set_aggregation("REPLACE"); + + tablet_schema->init_from_pb(tablet_schema_pb); + } + + void _create_mow_tablet() { + _tablet_schema = std::make_shared(); + _create_mow_tablet_schema(_tablet_schema); + + auto tablet_meta = std::make_shared(); + tablet_meta->_tablet_id = kTabletId; + tablet_meta->set_tablet_uid(_tablet_uid); + static_cast(tablet_meta->set_partition_id(kPartitionId)); + tablet_meta->_schema = _tablet_schema; + tablet_meta->_enable_unique_key_merge_on_write = true; + tablet_meta->_shard_id = 1; + tablet_meta->_schema_hash = kSchemaHash; + + auto tablet = + std::make_shared(*_engine, tablet_meta, _data_dir.get(), "ingest_test"); + Status st = tablet->init(); + ASSERT_TRUE(st.ok()) << st.to_string(); + + // Ensure the tablet directory exists so rowset writers can create segment files. + std::filesystem::create_directories(tablet->tablet_path()); + + auto& tablet_map = _engine->tablet_manager()->_get_tablet_map(kTabletId); + tablet_map[kTabletId] = tablet; + _tablet = tablet; + } + + RowsetSharedPtr _write_rowset(int64_t rowset_id, const PUniqueId& load_id, int num_segments, + int rows_per_segment) { + RowsetWriterContext writer_context; + RowsetId rs_id; + rs_id.init(rowset_id); + writer_context.rowset_id = rs_id; + writer_context.tablet_id = kTabletId; + writer_context.tablet_schema_hash = kSchemaHash; + writer_context.partition_id = kPartitionId; + writer_context.rowset_type = BETA_ROWSET; + writer_context.tablet_path = _tablet->tablet_path(); + writer_context.rowset_state = COMMITTED; + writer_context.tablet_schema = _tablet_schema; + writer_context.version.first = 1; + writer_context.version.second = 1; + writer_context.txn_id = kTxnId; + writer_context.load_id = load_id; + writer_context.tablet = _tablet; + writer_context.enable_unique_key_merge_on_write = true; + + // MOW context is required for merge-on-write rowsets so that segment + // metadata (primary key index) is generated for delete bitmap calculation. + auto rsids = std::make_shared(); + std::vector rowset_ptrs; + auto delete_bitmap = std::make_shared(kTabletId); + writer_context.mow_context = + std::make_shared(1, kTxnId, rsids, rowset_ptrs, delete_bitmap); + + auto res = RowsetFactory::create_rowset_writer(*_engine, writer_context, false); + EXPECT_TRUE(res.has_value()) << res.error(); + auto rowset_writer = std::move(res).value(); + + for (int seg = 0; seg < num_segments; ++seg) { + Block block = _tablet_schema->create_block(); + auto columns = std::move(block).mutate_columns(); + for (int rid = 0; rid < rows_per_segment; ++rid) { + // Use a small key space so the same keys appear in every segment. + // calc_delete_bitmap_between_segments then generates a non-empty + // delete bitmap keyed by the rowset id. + int32_t k1 = rid % 5; + int32_t v1 = k1 * 10; + int32_t seq = 0; + columns[0]->insert_data(reinterpret_cast(&k1), sizeof(k1)); + columns[1]->insert_data(reinterpret_cast(&v1), sizeof(v1)); + columns[2]->insert_data(reinterpret_cast(&seq), sizeof(seq)); + } + block.set_columns(std::move(columns)); + Status st = rowset_writer->add_block(&block); + EXPECT_TRUE(st.ok()) << st.to_string(); + st = rowset_writer->flush(); + EXPECT_TRUE(st.ok()) << st.to_string(); + } + + RowsetSharedPtr rowset; + Status st = rowset_writer->build(rowset); + EXPECT_TRUE(st.ok()) << st.to_string(); + return rowset; + } + + StorageEngine* _engine = nullptr; + std::unique_ptr _data_dir; + TabletSharedPtr _tablet; + TabletSchemaSPtr _tablet_schema; + TabletUid _tablet_uid {10001, 10002}; +}; + +// Verify that re-committing the same txn/load id with a different rowset id returns +// kAlreadyExist and does not overwrite the previously committed rowset or its delete bitmap. +TEST_F(BackendServiceIngestTest, CommitIngestedRowsetAlreadyExist) { + PUniqueId load_id; + load_id.set_hi(0); + load_id.set_lo(0); + + // First ingest commits R1. + auto rowset_r1 = _write_rowset(60001, load_id, 2, 1); + auto rowset_meta_r1 = rowset_r1->rowset_meta(); + auto guard_r1 = _engine->pending_local_rowsets().add(rowset_meta_r1->rowset_id()); + MonotonicStopWatch watch; + std::unordered_map elapsed_time_map; + auto result = commit_ingested_rowset(*_engine, _tablet, kTxnId, kPartitionId, rowset_meta_r1, + std::move(guard_r1), watch, elapsed_time_map); + ASSERT_EQ(result, IngestCommitResult::kCommitted); + + RowsetMetaSharedPtr committed_meta_r1(new RowsetMeta()); + Status st = RowsetMetaManager::get_rowset_meta(_data_dir->get_meta(), _tablet_uid, + rowset_meta_r1->rowset_id(), committed_meta_r1); + ASSERT_TRUE(st.ok()) << st.to_string(); + + // Capture the txn delete bitmap and rowset ids after R1 commit. + CommitTabletTxnInfoVec txn_info_vec_r1; + _engine->txn_manager()->get_all_commit_tablet_txn_info_by_tablet(*_tablet, &txn_info_vec_r1); + ASSERT_EQ(txn_info_vec_r1.size(), 1); + const auto& txn_info_r1 = txn_info_vec_r1[0]; + ASSERT_EQ(txn_info_r1.transaction_id, kTxnId); + ASSERT_EQ(txn_info_r1.partition_id, kPartitionId); + ASSERT_TRUE(txn_info_r1.delete_bitmap != nullptr); + const size_t r1_bitmap_count = txn_info_r1.delete_bitmap->get_delete_bitmap_count(); + const size_t r1_bitmap_cardinality = txn_info_r1.delete_bitmap->cardinality(); + const auto r1_rowset_ids = txn_info_r1.rowset_ids; + + // The same keys appear in both segments, so R1 must have generated a non-empty + // delete bitmap keyed by its own rowset id. The sentinel mark proves the + // multi-segment delete bitmap was calculated for R1. + ASSERT_GT(r1_bitmap_count, 0); + ASSERT_TRUE(txn_info_r1.delete_bitmap->contains( + {rowset_meta_r1->rowset_id(), DeleteBitmap::INVALID_SEGMENT_ID, + DeleteBitmap::TEMP_VERSION_COMMON}, + DeleteBitmap::ROWSET_SENTINEL_MARK)); + + // Second ingest with the same txn/load id but a different rowset id must be idempotent. + auto rowset_r2 = _write_rowset(60002, load_id, 2, 1); + auto rowset_meta_r2 = rowset_r2->rowset_meta(); + auto guard_r2 = _engine->pending_local_rowsets().add(rowset_meta_r2->rowset_id()); + elapsed_time_map.clear(); + result = commit_ingested_rowset(*_engine, _tablet, kTxnId, kPartitionId, rowset_meta_r2, + std::move(guard_r2), watch, elapsed_time_map); + ASSERT_EQ(result, IngestCommitResult::kAlreadyExist); + + // R1 is still the committed rowset; R2 must not have replaced it. + RowsetMetaSharedPtr committed_meta_r1_after(new RowsetMeta()); + st = RowsetMetaManager::get_rowset_meta(_data_dir->get_meta(), _tablet_uid, + rowset_meta_r1->rowset_id(), committed_meta_r1_after); + ASSERT_TRUE(st.ok()) << st.to_string(); + + RowsetMetaSharedPtr committed_meta_r2(new RowsetMeta()); + st = RowsetMetaManager::get_rowset_meta(_data_dir->get_meta(), _tablet_uid, + rowset_meta_r2->rowset_id(), committed_meta_r2); + ASSERT_FALSE(st.ok()); + + // The txn delete bitmap and rowset ids must be unchanged after kAlreadyExist. + CommitTabletTxnInfoVec txn_info_vec_r2; + _engine->txn_manager()->get_all_commit_tablet_txn_info_by_tablet(*_tablet, &txn_info_vec_r2); + ASSERT_EQ(txn_info_vec_r2.size(), 1); + const auto& txn_info_r2 = txn_info_vec_r2[0]; + ASSERT_EQ(txn_info_r2.transaction_id, kTxnId); + ASSERT_EQ(txn_info_r2.partition_id, kPartitionId); + ASSERT_TRUE(txn_info_r2.delete_bitmap != nullptr); + ASSERT_EQ(txn_info_r2.delete_bitmap->get_delete_bitmap_count(), r1_bitmap_count); + ASSERT_EQ(txn_info_r2.delete_bitmap->cardinality(), r1_bitmap_cardinality); + ASSERT_EQ(txn_info_r2.rowset_ids, r1_rowset_ids); + // R1's multi-segment sentinel mark must still be present; R2's must not appear. + ASSERT_TRUE(txn_info_r2.delete_bitmap->contains( + {rowset_meta_r1->rowset_id(), DeleteBitmap::INVALID_SEGMENT_ID, + DeleteBitmap::TEMP_VERSION_COMMON}, + DeleteBitmap::ROWSET_SENTINEL_MARK)); + ASSERT_FALSE(txn_info_r2.delete_bitmap->contains( + {rowset_meta_r2->rowset_id(), DeleteBitmap::INVALID_SEGMENT_ID, + DeleteBitmap::TEMP_VERSION_COMMON}, + DeleteBitmap::ROWSET_SENTINEL_MARK)); +} + +TEST_F(BackendServiceIngestTest, DeleteDownloadedFiles) { + auto tmp_dir = std::filesystem::path(kTestDir) / "delete_downloaded_files_test"; + std::filesystem::remove_all(tmp_dir); + std::filesystem::create_directories(tmp_dir); + + auto f1 = tmp_dir / "file1"; + auto f2 = tmp_dir / "file2"; + { + std::ofstream ofs(f1); + ofs << "data1"; + } + { + std::ofstream ofs(f2); + ofs << "data2"; + } + ASSERT_TRUE(std::filesystem::exists(f1)); + ASSERT_TRUE(std::filesystem::exists(f2)); + + std::vector files = {f1.string(), f2.string()}; + _delete_downloaded_files(files, "test cleanup", kTxnId); + ASSERT_FALSE(std::filesystem::exists(f1)); + ASSERT_FALSE(std::filesystem::exists(f2)); + + // Empty list should be a no-op. + _delete_downloaded_files({}, "empty cleanup", kTxnId); +} + +// Verify that fetch_from_peer rejects a rowset which claims to have segments but +// provides no file list. This is the reverse test for the empty-rowset fast path. +TEST_F(BackendServiceIngestTest, IngestBinlogFromPeerRejectsEmptyFilesWithSegments) { + PUniqueId pb_load_id; + pb_load_id.set_hi(0); + pb_load_id.set_lo(0); + + TUniqueId thrift_load_id; + thrift_load_id.hi = 0; + thrift_load_id.lo = 0; + + // Create a rowset with at least one segment. + auto rowset = _write_rowset(70001, pb_load_id, 2, 1); + auto rowset_meta = rowset->rowset_meta(); + std::string rowset_meta_str; + ASSERT_TRUE(rowset_meta->serialize(&rowset_meta_str)); + + TIngestBinlogRequest request; + request.__set_txn_id(kTxnId); + request.__set_partition_id(kPartitionId); + request.__set_local_tablet_id(kTabletId); + request.__set_load_id(thrift_load_id); + request.__set_fetch_from_peer(true); + request.__set_peer_host("127.0.0.1"); + request.__set_peer_http_port("8040"); + request.__set_peer_token("token"); + request.__set_rowset_meta(rowset_meta_str); + // files intentionally left empty while rowset_meta->num_segments() > 0 + + TStatus tstatus; + _ingest_binlog_from_peer(*_engine, request, _tablet, kTxnId, kPartitionId, tstatus); + ASSERT_EQ(tstatus.status_code, static_cast(TStatusCode::ANALYSIS_ERROR)) + << "expected ANALYSIS_ERROR when files is empty but num_segments > 0"; + ASSERT_FALSE(tstatus.error_msgs.empty()); +} + +} // namespace doris diff --git a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Syncer.groovy b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Syncer.groovy index a87744a86a387e..0bdb1eb5262958 100644 --- a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Syncer.groovy +++ b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Syncer.groovy @@ -37,6 +37,7 @@ import org.apache.doris.thrift.TGetSnapshotResult import org.apache.doris.thrift.TIngestBinlogRequest import org.apache.doris.thrift.TIngestBinlogResult import org.apache.doris.thrift.TNetworkAddress +import org.apache.doris.thrift.TReplicaDistributionInfo import org.apache.doris.thrift.TRestoreSnapshotResult import org.apache.doris.thrift.TStatus import org.apache.doris.thrift.TStatusCode @@ -577,7 +578,8 @@ class Syncer { } for (List row : backendInformation) { TNetworkAddress address = new TNetworkAddress(row[1] as String, row[3] as int) - BackendClientImpl client = new BackendClientImpl(address, row[4] as int) + int brpcPort = (row.size() > 5) ? (row[5] as int) : -1 + BackendClientImpl client = new BackendClientImpl(address, row[4] as int, brpcPort) clientsMap.put(row[0] as Long, client) } return clientsMap @@ -955,6 +957,166 @@ class Syncer { return true } + // Single-replica ingest binlog: the leader replica downloads the rowset and + // fans it out to followers inside BE. This matches the ccr-syncer + // handleSingleReplica path and requires the target table to have + // replication_num > 1. + Boolean ingestBinlogSingleReplica(long fakePartitionId = -1, long fakeVersion = -1) { + logger.info("Begin to ingest binlog with single replica download.") + + if (!context.metaIsValid()) { + logger.error("Meta data miss match, src: ${context.sourceTableMap}, target: ${context.targetTableMap}") + return false + } + + BinlogData binlogData = context.lastBinlog + if (binlogData == null || binlogData.tableRecords == null || binlogData.tableRecords.isEmpty()) { + logger.info("Skip ingest: lastBinlog has no tableRecords. lastBinlog=${binlogData}") + return true + } + + for (Entry tableInfo : context.sourceTableMap) { + String tableName = tableInfo.key + TableMeta srcTableMeta = tableInfo.value + if (!binlogData.tableRecords.containsKey(srcTableMeta.id)) { + continue + } + + PartitionRecords binlogRecords = binlogData.tableRecords.get(srcTableMeta.id) + TableMeta tarTableMeta = context.targetTableMap.get(tableName) + + Iterator sourcePartitionIter = srcTableMeta.partitionMap.iterator() + Iterator targetPartitionIter = tarTableMeta.partitionMap.iterator() + + while (sourcePartitionIter.hasNext()) { + Entry srcPartition = sourcePartitionIter.next() + Entry tarPartition = targetPartitionIter.next() + if (!binlogRecords.contains(srcPartition.key)) { + continue + } + + for (PartitionData partitionRecord : binlogRecords.partitionRecords) { + if (partitionRecord.partitionId != srcPartition.key) { + continue + } + + long txnId = partitionRecord.stid == -1 ? context.txnId : context.sourceToTargetSubTxnId.get(partitionRecord.stid) + long partitionId = fakePartitionId == -1 ? tarPartition.key : fakePartitionId + long version = fakeVersion == -1 ? partitionRecord.version : fakeVersion + + Iterator srcTabletIter = srcPartition.value.tabletMeta.iterator() + Iterator tarTabletIter = tarPartition.value.tabletMeta.iterator() + while (srcTabletIter.hasNext()) { + Entry srcTabletMap = srcTabletIter.next() + Entry tarTabletMap = tarTabletIter.next() + TabletMeta srcTabletMeta = srcTabletMap.value + TabletMeta tarTabletMeta = tarTabletMap.value + + if (tarTabletMeta.replicas.size() <= 1) { + logger.error("Single replica ingest requires target tablet has more than 1 replica, tabletId=${tarTabletMap.key}") + return false + } + + // Pick leader by tablet id hash to match ccr-syncer logic. + int leaderIdx = (int) (tarTabletMap.key % tarTabletMeta.replicas.size()) + Iterator tarReplicaIter = tarTabletMeta.replicas.iterator() + int idx = 0 + long leaderBackendId = -1 + List followerReplicas = new ArrayList() + List allBackendIds = new ArrayList() + while (tarReplicaIter.hasNext()) { + Entry tarReplicaMap = tarReplicaIter.next() + if (idx == leaderIdx) { + leaderBackendId = tarReplicaMap.value + } else { + BackendClientImpl followerClient = context.targetBackendClients.get(tarReplicaMap.value) + if (followerClient == null) { + logger.error("Can't find follower target tabletId-${tarTabletMap.key} -> beId-${tarReplicaMap.value}") + return false + } + if (followerClient.address.port <= 0) { + logger.error("Follower backend be port is invalid, beId=${tarReplicaMap.value}") + return false + } + TReplicaDistributionInfo info = new TReplicaDistributionInfo() + info.setBackendId(tarReplicaMap.value) + info.setHost(followerClient.address.hostname) + info.setBePort(followerClient.address.port) + followerReplicas.add(info) + } + allBackendIds.add(tarReplicaMap.value) + idx++ + } + + BackendClientImpl leaderClient = context.targetBackendClients.get(leaderBackendId) + if (leaderClient == null) { + logger.error("Can't find leader target tabletId-${tarTabletMap.key} -> beId=${leaderBackendId}") + return false + } + + // Pick source replica by tablet id hash. + int srcIdx = (int) (srcTabletMap.key % srcTabletMeta.replicas.size()) + Iterator srcReplicaIter = srcTabletMeta.replicas.iterator() + int sIdx = 0 + long srcBackendId = -1 + while (srcReplicaIter.hasNext()) { + Entry srcReplicaMap = srcReplicaIter.next() + if (sIdx == srcIdx) { + srcBackendId = srcReplicaMap.value + break + } + sIdx++ + } + BackendClientImpl srcClient = context.sourceBackendClients.get(srcBackendId) + if (srcClient == null) { + logger.error("Can't find src tabletId-${srcTabletMap.key} -> beId-${srcBackendId}") + return false + } + + tarPartition.value.version = srcPartition.value.version + + TIngestBinlogRequest request = new TIngestBinlogRequest() + TUniqueId uid = new TUniqueId(-1, -1) + request.setTxnId(txnId) + request.setRemoteTabletId(srcTabletMap.key) + request.setBinlogVersion(version) + request.setRemoteHost(srcClient.address.hostname) + request.setRemotePort(srcClient.httpPort.toString()) + request.setPartitionId(partitionId) + request.setLocalTabletId(tarTabletMap.key) + request.setLoadId(uid) + request.setSingleReplicaDownload(true) + request.setFollowerReplicas(followerReplicas) + logger.info("single replica request -> ${request}") + TIngestBinlogResult result = leaderClient.client.ingestBinlog(request) + if (!checkIngestBinlog(result)) { + logger.error("Single replica ingest binlog error! result: ${result}") + return false + } + if (!result.isSetSuccessReplicaBackendIds()) { + logger.error("Single replica ingest result has no success_replica_backend_ids, old BE fallback") + return false + } + + Set successBackendIds = new HashSet(result.getSuccessReplicaBackendIds()) + for (TReplicaDistributionInfo follower : followerReplicas) { + long backendId = follower.getBackendId() + if (!successBackendIds.contains(backendId)) { + logger.error("Single replica ingest follower failed, backendId=${backendId}, result=${result}") + return false + } + } + + for (long backendId : allBackendIds) { + addCommitInfo(tarTabletMap.key, backendId) + } + } + } + } + } + return true + } + Boolean commitTxn() { logger.info("Commit transaction to target cluster ${context.config.feTargetThriftNetworkAddress}, txnId: ${context.txnId}") FrontendClientImpl clientImpl = context.getTargetFrontClient() diff --git a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/client/BackendClientImpl.groovy b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/client/BackendClientImpl.groovy index 5695baa729d926..f9dbb6e02c2d81 100644 --- a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/client/BackendClientImpl.groovy +++ b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/client/BackendClientImpl.groovy @@ -28,11 +28,13 @@ class BackendClientImpl { public TNetworkAddress address public int httpPort + public int brpcPort public BackendService.Client client - BackendClientImpl(TNetworkAddress address, int httpPort) throws TTransportException { + BackendClientImpl(TNetworkAddress address, int httpPort, int brpcPort = -1) throws TTransportException { this.address = address this.httpPort = httpPort + this.brpcPort = brpcPort this.tSocket = new TSocket(address.hostname, address.port) this.client = new BackendService.Client(new TBinaryProtocol(this.tSocket)) this.tSocket.open() diff --git a/regression-test/suites/ccr_syncer_p0/test_single_replica_ingest_binlog.groovy b/regression-test/suites/ccr_syncer_p0/test_single_replica_ingest_binlog.groovy new file mode 100644 index 00000000000000..321d4ba6e9b664 --- /dev/null +++ b/regression-test/suites/ccr_syncer_p0/test_single_replica_ingest_binlog.groovy @@ -0,0 +1,406 @@ +// 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. + +import org.apache.doris.regression.util.DebugPoint +import org.apache.doris.regression.util.Http +import org.apache.doris.regression.util.NodeType + +suite("test_single_replica_ingest_binlog") { + + def syncer = getSyncer() + if (!syncer.checkEnableFeatureBinlog()) { + logger.info("fe enable_feature_binlog is false, skip case test_single_replica_ingest_binlog") + return + } + + def tableName = "tbl_single_replica_ingest_binlog" + def insert_num = 5 + + sql "DROP TABLE IF EXISTS ${tableName}" + sql """ + CREATE TABLE IF NOT EXISTS ${tableName} + ( + `test` INT, + `id` INT + ) + ENGINE=OLAP + UNIQUE KEY(`test`, `id`) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + sql """ALTER TABLE ${tableName} set ("binlog.enable" = "true")""" + + target_sql "DROP TABLE IF EXISTS ${tableName}" + target_sql """ + CREATE TABLE IF NOT EXISTS ${tableName} + ( + `test` INT, + `id` INT + ) + ENGINE=OLAP + UNIQUE KEY(`test`, `id`) + DISTRIBUTED BY HASH(id) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + assertTrue(syncer.getTargetMeta("${tableName}")) + + logger.info("=== Test 1: Single replica ingest binlog case ===") + for (int index = 0; index < insert_num; index++) { + sql """INSERT INTO ${tableName} VALUES (1, ${index})""" + assertTrue(syncer.getBinlog("${tableName}")) + assertTrue(syncer.beginTxn("${tableName}")) + assertTrue(syncer.getBackendClients()) + assertTrue(syncer.ingestBinlogSingleReplica()) + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + } + + target_sql "sync" + def res = target_sql """SELECT * FROM ${tableName} WHERE test=1 ORDER BY id""" + assertEquals(res.size(), insert_num) + + logger.info("=== Test 2: Idempotent re-ingest for the same txn ===") + sql """INSERT INTO ${tableName} VALUES (2, 0)""" + assertTrue(syncer.getBinlog("${tableName}")) + assertTrue(syncer.beginTxn("${tableName}")) + assertTrue(syncer.getBackendClients()) + // First ingest should succeed. + assertTrue(syncer.ingestBinlogSingleReplica()) + // Re-ingest the same txn with the same load_id should be idempotent. + assertTrue(syncer.ingestBinlogSingleReplica()) + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + + target_sql "sync" + res = target_sql """SELECT * FROM ${tableName} WHERE test=2""" + assertEquals(res.size(), 1) + + logger.info("=== Test 3: Multi-bucket table with empty rowsets ===") + def bucketTableName = "tbl_single_replica_ingest_binlog_buckets" + sql "DROP TABLE IF EXISTS ${bucketTableName}" + sql """ + CREATE TABLE IF NOT EXISTS ${bucketTableName} + ( + `k` INT, + `v` INT + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 3 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + sql """ALTER TABLE ${bucketTableName} set ("binlog.enable" = "true")""" + + target_sql "DROP TABLE IF EXISTS ${bucketTableName}" + target_sql """ + CREATE TABLE IF NOT EXISTS ${bucketTableName} + ( + `k` INT, + `v` INT + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 3 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + assertTrue(syncer.getTargetMeta("${bucketTableName}")) + + // Insert values that hash to only some buckets, leaving other tablets with empty rowsets. + sql """INSERT INTO ${bucketTableName} VALUES (1, 10), (2, 20)""" + assertTrue(syncer.getBinlog("${bucketTableName}")) + assertTrue(syncer.beginTxn("${bucketTableName}")) + assertTrue(syncer.getBackendClients()) + assertTrue(syncer.ingestBinlogSingleReplica()) + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + + target_sql "sync" + res = target_sql """SELECT * FROM ${bucketTableName} ORDER BY k""" + assertEquals(res.size(), 2) + + logger.info("=== Test 4: MOW table idempotent re-ingest (delete bitmap must not be overwritten) ===") + def mowTableName = "tbl_single_replica_ingest_binlog_mow" + sql "DROP TABLE IF EXISTS ${mowTableName}" + sql """ + CREATE TABLE IF NOT EXISTS ${mowTableName} + ( + `k` INT, + `v` STRING + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "enable_unique_key_merge_on_write" = "true", + "replication_allocation" = "tag.location.default: 3" + ) + """ + sql """ALTER TABLE ${mowTableName} set ("binlog.enable" = "true")""" + + target_sql "DROP TABLE IF EXISTS ${mowTableName}" + target_sql """ + CREATE TABLE IF NOT EXISTS ${mowTableName} + ( + `k` INT, + `v` STRING + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "enable_unique_key_merge_on_write" = "true", + "replication_allocation" = "tag.location.default: 3" + ) + """ + assertTrue(syncer.getTargetMeta("${mowTableName}")) + + // Phase 1: establish baseline and publish it. + sql """INSERT INTO ${mowTableName} VALUES (1, '10'), (2, '20')""" + assertTrue(syncer.getBinlog("${mowTableName}")) + assertTrue(syncer.beginTxn("${mowTableName}")) + assertTrue(syncer.getBackendClients()) + assertTrue(syncer.ingestBinlogSingleReplica()) + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + + target_sql "sync" + res = target_sql """SELECT k, v FROM ${mowTableName} ORDER BY k""" + assertEquals(res.size(), 2) + assertEquals(res[0][0], 1) + assertEquals(res[0][1], '10') + assertEquals(res[1][0], 2) + assertEquals(res[1][1], '20') + + // Phase 2: update an existing key and then ingest the same txn twice. + // A reliable multi-segment source rowset is hard to construct in regression + // (MOW deduplicates keys in the memtable and debug points are unavailable in + // release builds), so the strict rowset-id bitmap coverage is deferred to a + // BE unit test. Here we still verify the idempotent ingest flow and data. + sql """INSERT INTO ${mowTableName} SELECT 2, repeat('x', 1000000) FROM numbers("number" = "100")""" + + assertTrue(syncer.getBinlog("${mowTableName}")) + assertTrue(syncer.beginTxn("${mowTableName}")) + assertTrue(syncer.getBackendClients()) + assertTrue(syncer.ingestBinlogSingleReplica()) + assertTrue(syncer.ingestBinlogSingleReplica()) + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + + target_sql "sync" + res = target_sql """SELECT k, v FROM ${mowTableName} ORDER BY k""" + assertEquals(res.size(), 2) + assertEquals(res[0][0], 1) + assertEquals(res[0][1], '10') + assertEquals(res[1][0], 2) + // The large value for k=2 should be visible; if the bitmap were overwritten by + // a non-existent R2 rowset id, the row would be hidden or the query would fail. + assertEquals(res[1][1].length(), 1000000) + + logger.info("=== Test 5: Legacy multi-replica ingest binlog path regression ===") + def legacyTableName = "tbl_single_replica_ingest_binlog_legacy" + sql "DROP TABLE IF EXISTS ${legacyTableName}" + sql """ + CREATE TABLE IF NOT EXISTS ${legacyTableName} + ( + `k` INT, + `v` INT + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + sql """ALTER TABLE ${legacyTableName} set ("binlog.enable" = "true")""" + + target_sql "DROP TABLE IF EXISTS ${legacyTableName}" + target_sql """ + CREATE TABLE IF NOT EXISTS ${legacyTableName} + ( + `k` INT, + `v` INT + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + assertTrue(syncer.getTargetMeta("${legacyTableName}")) + + sql """INSERT INTO ${legacyTableName} VALUES (1, 10), (2, 20)""" + assertTrue(syncer.getBinlog("${legacyTableName}")) + assertTrue(syncer.beginTxn("${legacyTableName}")) + assertTrue(syncer.getBackendClients()) + assertTrue(syncer.ingestBinlog()) + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + + target_sql "sync" + res = target_sql """SELECT * FROM ${legacyTableName} ORDER BY k""" + assertEquals(res.size(), 2) + assertEquals(res[0][0], 1) + assertEquals(res[0][1], 10) + assertEquals(res[1][0], 2) + assertEquals(res[1][1], 20) + + logger.info("=== Test 6: Follower failure and retry with kAlreadyExist ===") + def retryTableName = "tbl_single_replica_ingest_binlog_follower_retry" + sql "DROP TABLE IF EXISTS ${retryTableName}" + sql """ + CREATE TABLE IF NOT EXISTS ${retryTableName} + ( + `k` INT, + `v` INT + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + sql """ALTER TABLE ${retryTableName} set ("binlog.enable" = "true")""" + + target_sql "DROP TABLE IF EXISTS ${retryTableName}" + target_sql """ + CREATE TABLE IF NOT EXISTS ${retryTableName} + ( + `k` INT, + `v` INT + ) + ENGINE=OLAP + UNIQUE KEY(`k`) + DISTRIBUTED BY HASH(k) BUCKETS 1 + PROPERTIES ( + "replication_allocation" = "tag.location.default: 3" + ) + """ + assertTrue(syncer.getTargetMeta("${retryTableName}")) + + sql """INSERT INTO ${retryTableName} VALUES (1, 10), (2, 20)""" + assertTrue(syncer.getBinlog("${retryTableName}")) + assertTrue(syncer.beginTxn("${retryTableName}")) + assertTrue(syncer.getBackendClients()) + + // Pick the leader backend (where the debug point is evaluated) and a follower to fail. + def tarTableMeta = syncer.context.targetTableMap.get(retryTableName) + def tarPartitionMeta = tarTableMeta.partitionMap.values().iterator().next() + def tarTabletEntry = tarPartitionMeta.tabletMeta.entrySet().iterator().next() + def tabletId = tarTabletEntry.key + def tarTabletMeta = tarTabletEntry.value + def leaderIdx = (int) (tabletId % tarTabletMeta.replicas.size()) + def leaderBackendId = -1L + def followerBackendId = -1L + def successFollowerBackendId = -1L + tarTabletMeta.replicas.eachWithIndex { entry, idx -> + if (idx == leaderIdx) { + leaderBackendId = entry.value + } else if (followerBackendId == -1L) { + followerBackendId = entry.value + } else { + successFollowerBackendId = entry.value + } + } + assertTrue(leaderBackendId != -1L, "should find leader replica") + assertTrue(followerBackendId != -1L, "should find a follower replica to fail") + assertTrue(successFollowerBackendId != -1L, "should find a follower replica that succeeds") + + def leaderClient = syncer.context.targetBackendClients.get(leaderBackendId) + assertTrue(leaderClient != null, "should find leader backend client") + + def followerClient = syncer.context.targetBackendClients.get(followerBackendId) + assertTrue(followerClient != null, "should find follower backend client") + + def successFollowerClient = syncer.context.targetBackendClients.get(successFollowerBackendId) + assertTrue(successFollowerClient != null, "should find success follower backend client") + + def readMetric = { host, httpPort, metricName -> + def url = "http://${host}:${httpPort}/metrics" + def text = Http.GET(url, false, false) + def m = text =~ /(?m)^${metricName}\s+(\d+)$/ + return m ? Long.parseLong(m[0][1]) : -1L + } + def metricName = "doris_be_binlog_ingest_redundant_rowset_cleanup_success_total" + + // The debug point runs in the leader's distribution path and forces the + // specified follower backend id to fail. enable_debug_points is a static + // (non-dynamic) BE config, so the cluster must be started with + // enable_debug_points=true in be.conf; the test relies on that startup + // configuration rather than trying to toggle it at runtime. + try { + DebugPoint.enableDebugPoint(leaderClient.address.hostname, leaderClient.httpPort, + NodeType.BE, "ingest_binlog.follower.force_fail", + ["backend_id": String.valueOf(followerBackendId)]) + + // First ingest should fail because the follower is forced to fail. + def firstIngestOk = syncer.ingestBinlogSingleReplica() + assert !firstIngestOk : "first ingest should fail when follower is forced down" + } finally { + DebugPoint.disableDebugPoint(leaderClient.address.hostname, leaderClient.httpPort, + NodeType.BE, "ingest_binlog.follower.force_fail") + } + + // Retry: the leader already committed the rowset, so this attempt should hit + // kAlreadyExist and still fan out to followers. The follower that was forced to fail + // in the first attempt has no committed rowset, so it will not hit kAlreadyExist. + // The follower that succeeded in the first attempt will hit kAlreadyExist and delete + // redundant peer files, so we check its metric. + def leaderMetricBefore = readMetric(leaderClient.address.hostname, leaderClient.httpPort, metricName) + def successFollowerMetricBefore = readMetric(successFollowerClient.address.hostname, successFollowerClient.httpPort, metricName) + logger.info("redundant files deleted metric before retry: leader=${leaderMetricBefore}, successFollower=${successFollowerMetricBefore}") + assertTrue(leaderMetricBefore != -1, "leader metric ${metricName} should exist") + assertTrue(successFollowerMetricBefore != -1, "success follower metric ${metricName} should exist") + + assertTrue(syncer.ingestBinlogSingleReplica()) + + def leaderMetricAfter = readMetric(leaderClient.address.hostname, leaderClient.httpPort, metricName) + def successFollowerMetricAfter = readMetric(successFollowerClient.address.hostname, successFollowerClient.httpPort, metricName) + logger.info("redundant files deleted metric after retry: leader=${leaderMetricAfter}, successFollower=${successFollowerMetricAfter}") + assertTrue(leaderMetricAfter > leaderMetricBefore, + "leader should delete redundant rowset files after kAlreadyExist retry") + assertTrue(successFollowerMetricAfter > successFollowerMetricBefore, + "follower that succeeded first should delete redundant peer files after kAlreadyExist retry") + + assertTrue(syncer.commitTxn()) + assertTrue(syncer.checkTargetVersion()) + syncer.closeBackendClients() + + target_sql "sync" + res = target_sql """SELECT * FROM ${retryTableName} ORDER BY k""" + assertEquals(res.size(), 2) + assertEquals(res[0][0], 1) + assertEquals(res[0][1], 10) + assertEquals(res[1][0], 2) + assertEquals(res[1][1], 20) +}