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/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/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/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/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/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 { 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) +}