From fba999804ca1c12d21e57d8b61cb31be7988844a Mon Sep 17 00:00:00 2001 From: honestmanxin Date: Mon, 26 Oct 2026 19:45:32 +0800 Subject: [PATCH] [fix](load) Exclude disable-load backends from tablet writes --- be/src/exec/sink/load_stream_map_pool.cpp | 60 ++++++ be/src/exec/sink/load_stream_map_pool.h | 15 ++ be/src/exec/sink/writer/vtablet_writer.cpp | 190 +++++++++++++---- be/src/exec/sink/writer/vtablet_writer.h | 14 +- be/src/exec/sink/writer/vtablet_writer_v2.cpp | 179 +++++++++++++--- be/src/exec/sink/writer/vtablet_writer_v2.h | 23 ++- .../exec/sink/vtablet_writer_failure_test.cpp | 192 ++++++++++++++++++ be/test/exec/sink/vtablet_writer_v2_test.cpp | 190 ++++++++++++++++- .../java/org/apache/doris/catalog/Tablet.java | 9 +- .../doris/cloud/catalog/CloudTablet.java | 15 ++ .../apache/doris/planner/OlapTableSink.java | 12 +- .../java/org/apache/doris/qe/Coordinator.java | 4 + .../java/org/apache/doris/qe/LoadContext.java | 4 + .../doris/service/FrontendServiceImpl.java | 36 +++- .../doris/system/SystemInfoService.java | 8 + .../transaction/DatabaseTransactionMgr.java | 7 +- .../doris/transaction/ErrorTabletInfo.java | 13 ++ .../doris/transaction/TabletCommitInfo.java | 1 + .../org/apache/doris/catalog/TabletTest.java | 12 ++ .../doris/planner/OlapTableSinkTest.java | 99 +++++++++ .../org/apache/doris/qe/LoadContextTest.java | 81 ++++++++ .../service/FrontendServiceImplTest.java | 87 ++++++++ .../DatabaseTransactionMgrTest.java | 22 ++ ...sable_load_selected_replica_failure.groovy | 144 +++++++++++++ 24 files changed, 1315 insertions(+), 102 deletions(-) create mode 100644 be/test/exec/sink/vtablet_writer_failure_test.cpp create mode 100644 fe/fe-core/src/test/java/org/apache/doris/qe/LoadContextTest.java create mode 100644 regression-test/suites/fault_injection_p0/test_disable_load_selected_replica_failure.groovy diff --git a/be/src/exec/sink/load_stream_map_pool.cpp b/be/src/exec/sink/load_stream_map_pool.cpp index 6d87fc19840c14..692da4dca6d07a 100644 --- a/be/src/exec/sink/load_stream_map_pool.cpp +++ b/be/src/exec/sink/load_stream_map_pool.cpp @@ -17,6 +17,8 @@ #include "exec/sink/load_stream_map_pool.h" +#include + #include "util/debug_points.h" namespace doris { @@ -99,6 +101,64 @@ void LoadStreamMap::save_tablets_to_commit(int64_t dst_id, } } +void LoadStreamMap::append_partition_involvement_commit_infos( + std::vector& tablet_commit_infos) { + constexpr int64_t PARTITION_INVOLVEMENT_BACKEND_ID = -1; + constexpr int64_t DUMMY_TABLET_ID = 0; + std::unordered_map partition_to_tablet; + { + std::lock_guard lock(_tablets_to_commit_mutex); + for (const auto& [_, tablets] : _tablets_to_commit) { + for (const auto& [tablet_id, tablet] : tablets) { + if (tablet_id != DUMMY_TABLET_ID) { + partition_to_tablet.try_emplace(tablet.partition_id(), tablet_id); + } + } + } + } + for (const auto& [_, tablet_id] : partition_to_tablet) { + TTabletCommitInfo commit_info; + commit_info.__set_tabletId(tablet_id); + commit_info.__set_backendId(PARTITION_INVOLVEMENT_BACKEND_ID); + tablet_commit_infos.emplace_back(std::move(commit_info)); + } +} + +Status LoadStreamMap::merge_tablet_replica_info(int64_t tablet_id, + const TabletReplicaInfo& replica_info) { + std::lock_guard lock(_mutex); + auto [it, inserted] = _tablet_replica_infos.emplace(tablet_id, replica_info); + if (!inserted) { + auto& existing = it->second; + if (existing.selected_replicas_num != replica_info.selected_replicas_num) { + return Status::InternalError("conflicting replica information for tablet {} in load {}", + tablet_id, _load_id.to_string()); + } + // Dynamic min_load_replica_num can change between concurrent createPartition replies. + // Replica allocation can also be refreshed while cached destinations stay unchanged. + // Keep the strictest required count and the smallest configured failure allowance. + int configured_max_failed = + std::min(existing.total_replicas_num - existing.load_required_replicas_num, + replica_info.total_replicas_num - replica_info.load_required_replicas_num); + existing.load_required_replicas_num = std::max(existing.load_required_replicas_num, + replica_info.load_required_replicas_num); + existing.total_replicas_num = existing.load_required_replicas_num + configured_max_failed; + existing.version_gap_backends.insert(replica_info.version_gap_backends.begin(), + replica_info.version_gap_backends.end()); + } + return Status::OK(); +} + +bool LoadStreamMap::get_tablet_replica_info(int64_t tablet_id, TabletReplicaInfo* replica_info) { + std::lock_guard lock(_mutex); + auto it = _tablet_replica_infos.find(tablet_id); + if (it == _tablet_replica_infos.end()) { + return false; + } + *replica_info = it->second; + return true; +} + bool LoadStreamMap::release() { int num_use = --_use_cnt; if (num_use == 0) { diff --git a/be/src/exec/sink/load_stream_map_pool.h b/be/src/exec/sink/load_stream_map_pool.h index 131d1d1d88b2b7..11dbd0800ea613 100644 --- a/be/src/exec/sink/load_stream_map_pool.h +++ b/be/src/exec/sink/load_stream_map_pool.h @@ -70,6 +70,13 @@ class LoadStreamMapPool; class LoadStreamMap { public: + struct TabletReplicaInfo { + int total_replicas_num; + int load_required_replicas_num; + int selected_replicas_num; + std::unordered_set version_gap_backends; + }; + LoadStreamMap(UniqueId load_id, int64_t src_id, int num_streams, int num_use, LoadStreamMapPool* pool); @@ -85,6 +92,13 @@ class LoadStreamMap { void save_tablets_to_commit(int64_t dst_id, const std::vector& tablets_to_commit); + void append_partition_involvement_commit_infos( + std::vector& tablet_commit_infos); + + Status merge_tablet_replica_info(int64_t tablet_id, const TabletReplicaInfo& replica_info); + + bool get_tablet_replica_info(int64_t tablet_id, TabletReplicaInfo* replica_info); + void save_segments_for_tablet(const std::unordered_map& segments_for_tablet) { _segments_for_tablet.insert(segments_for_tablet.cbegin(), segments_for_tablet.cend()); } @@ -117,6 +131,7 @@ class LoadStreamMap { std::atomic _num_incremental_streams; std::mutex _mutex; std::unordered_map> _streams_for_node; + std::unordered_map _tablet_replica_infos; LoadStreamMapPool* _pool = nullptr; std::shared_ptr _tablet_schema_for_index; std::shared_ptr _enable_unique_mow_for_index; diff --git a/be/src/exec/sink/writer/vtablet_writer.cpp b/be/src/exec/sink/writer/vtablet_writer.cpp index fa2bff788c1ea2..6225eabfeaa326 100644 --- a/be/src/exec/sink/writer/vtablet_writer.cpp +++ b/be/src/exec/sink/writer/vtablet_writer.cpp @@ -46,6 +46,7 @@ #include #include "cloud/config.h" +#include "common/check.h" #include "common/compiler_util.h" // IWYU pragma: keep #include "common/config.h" #include "common/logging.h" @@ -92,6 +93,8 @@ namespace doris { class TExpr; +static constexpr int64_t V1_DUMMY_TABLET_ID = 0; + bvar::Adder g_sink_write_bytes; bvar::PerSecond> g_sink_write_bytes_per_second("sink_throughput_byte", &g_sink_write_bytes, 60); @@ -136,6 +139,17 @@ Status IndexChannel::init(RuntimeState* state, const std::vector(tablet_locations->node_ids.size()); + int load_required_replicas_num = _load_required_replicas_num(tablet.tablet_id); + if (selected_replicas_num < load_required_replicas_num) { + // Returning OK here lets a tablet with no viable quorum produce no commit info, + // so a mixed-partition transaction may omit that partition from FE commit checks. + return Status::InternalError( + "tablet {} in partition {} has {} selected replicas, fewer than required {}, " + "load_id={}", + tablet.tablet_id, tablet.partition_id, selected_replicas_num, + load_required_replicas_num, print_id(_parent->_load_id)); + } std::vector> channels; // For tablet, deal with its' all replica (in some node). for (auto& replica_node_id : tablet_locations->node_ids) { @@ -174,7 +188,11 @@ Status IndexChannel::init(RuntimeState* state, const std::vectorchannel_info() << " tablet_id: " << tablet_id << " as failed, err: " << err; auto node_id = node_channel->node_id(); + if (tablet_id != -1) { + std::lock_guard l(_fail_lock); + if (!_parent->_tablet_replica_info.contains(tablet_id)) { + // A receiver closes the union of partitions touched by all senders, so its final + // closer can receive another sender's tablet error without that tablet's policy. + // Shared-nothing FE combines this error marker with commit infos from every sender + // and validates the tablet quorum globally. Cloud commit has no such aggregation. + _failed_channels[tablet_id].insert(node_id); + _failed_channels_msgs.emplace(tablet_id, err + ", host: " + node_channel->host()); + if (config::is_cloud_mode()) { + _intolerable_failure_status = Status::InternalError( + "received failure for nonlocal tablet {} from backend {}: {}", tablet_id, + node_id, err); + } + return; + } + _failed_channels[tablet_id].insert(node_id); + _failed_channels_msgs.emplace(tablet_id, err + ", host: " + node_channel->host()); + if (_failed_channels[tablet_id].size() > _max_failed_replicas(tablet_id)) { + _intolerable_failure_status = Status::Error( + _failed_channels_msgs[tablet_id]); + } + return; + } + const auto& it = _tablets_by_channel.find(node_id); if (it == _tablets_by_channel.end()) { + std::lock_guard l(_fail_lock); + // A bootstrap-only final sender has no real tablet bookkeeping for this backend. An + // overall RPC/transport failure can still belong to another sender's real tablet. + _intolerable_failure_status = Status::InternalError( + "channel failure from backend {} has no local real tablet: {}", node_id, err); return; } - { - std::lock_guard l(_fail_lock); - if (tablet_id == -1) { - for (const auto the_tablet_id : it->second) { - _failed_channels[the_tablet_id].insert(node_id); - _failed_channels_msgs.emplace(the_tablet_id, - err + ", host: " + node_channel->host()); - if (_failed_channels[the_tablet_id].size() > _max_failed_replicas(the_tablet_id)) { - _intolerable_failure_status = Status::Error( - _failed_channels_msgs[the_tablet_id]); - } - } - } else { - _failed_channels[tablet_id].insert(node_id); - _failed_channels_msgs.emplace(tablet_id, err + ", host: " + node_channel->host()); - if (_failed_channels[tablet_id].size() > _max_failed_replicas(tablet_id)) { - _intolerable_failure_status = Status::Error( - _failed_channels_msgs[tablet_id]); - } + std::lock_guard l(_fail_lock); + for (const auto the_tablet_id : it->second) { + _failed_channels[the_tablet_id].insert(node_id); + _failed_channels_msgs.emplace(the_tablet_id, err + ", host: " + node_channel->host()); + if (_failed_channels[the_tablet_id].size() > _max_failed_replicas(the_tablet_id)) { + _intolerable_failure_status = Status::Error( + _failed_channels_msgs[the_tablet_id]); } } } int IndexChannel::_max_failed_replicas(int64_t tablet_id) { - auto [total_replicas_num, load_required_replicas_num] = - _parent->_tablet_replica_info[tablet_id]; - int max_failed_replicas = total_replicas_num == 0 - ? (_parent->_num_replicas - 1) / 2 - : total_replicas_num - load_required_replicas_num; - return max_failed_replicas; + DORIS_CHECK(_parent->_tablet_replica_info.contains(tablet_id)); + const auto& replica_info = _parent->_tablet_replica_info.at(tablet_id); + int configured_max_failed = + replica_info.total_replicas_num == 0 + ? (_parent->_num_replicas - 1) / 2 + : replica_info.total_replicas_num - replica_info.load_required_replicas_num; + DORIS_CHECK_GE(configured_max_failed, 0); + // A replica omitted from this load's locations cannot absorb a later channel failure. + // For total=3, required=1, selected=1, the selected target must tolerate zero failures. + int selected_max_failed = + replica_info.selected_replicas_num - replica_info.load_required_replicas_num; + DORIS_CHECK_GE(selected_max_failed, 0); + return std::min(configured_max_failed, selected_max_failed); } int IndexChannel::_load_required_replicas_num(int64_t tablet_id) { - auto [total_replicas_num, load_required_replicas_num] = - _parent->_tablet_replica_info[tablet_id]; - if (total_replicas_num == 0) { + DORIS_CHECK(_parent->_tablet_replica_info.contains(tablet_id)); + const auto& replica_info = _parent->_tablet_replica_info.at(tablet_id); + if (replica_info.total_replicas_num == 0) { return (_parent->_num_replicas + 1) / 2; } - return load_required_replicas_num; + return replica_info.load_required_replicas_num; } Status IndexChannel::check_intolerable_failure() { @@ -1786,7 +1829,7 @@ Status VTabletWriter::_init(RuntimeState* state, RuntimeProfile* profile) { tablet_with_partition.partition_id = part->id; tablet_with_partition.tablet_id = tablet; tablets.emplace_back(std::move(tablet_with_partition)); - _build_tablet_replica_info(tablet, part); + RETURN_IF_ERROR(_build_tablet_replica_info(tablet, part)); } } if (tablets.empty() && !_vpartition->is_auto_partition()) { @@ -1819,7 +1862,7 @@ Status VTabletWriter::_incremental_open_node_channel( tablet_with_partition.partition_id = part->id; tablet_with_partition.tablet_id = tablet; tablets.emplace_back(std::move(tablet_with_partition)); - _build_tablet_replica_info(tablet, part); + RETURN_IF_ERROR(_build_tablet_replica_info(tablet, part)); } DCHECK(!tablets.empty()) << "incremental open got nothing!"; } @@ -1856,25 +1899,91 @@ Status VTabletWriter::_incremental_open_node_channel( return Status::OK(); } -void VTabletWriter::_build_tablet_replica_info(const int64_t tablet_id, - VOlapTablePartition* partition) { +Status VTabletWriter::_build_tablet_replica_info(const int64_t tablet_id, + VOlapTablePartition* partition) { + const auto* tablet_location = _location->find_tablet(tablet_id); + if (tablet_location == nullptr) { + return Status::InternalError("unknown tablet location, tablet_id={}", tablet_id); + } + int selected_replicas_num = cast_set(tablet_location->node_ids.size()); + if (selected_replicas_num == 0) { + return Status::InternalError("tablet {} has no selected replicas", tablet_id); + } + if (tablet_id == V1_DUMMY_TABLET_ID) [[unlikely]] { + if (selected_replicas_num != 1) { + return Status::InternalError( + "bootstrap tablet {} must have exactly one selected replica, actual {}", + tablet_id, selected_replicas_num); + } + _tablet_replica_info.emplace(tablet_id, TabletReplicaInfo {.total_replicas_num = 1, + .load_required_replicas_num = 1, + .selected_replicas_num = 1}); + return Status::OK(); + } if (partition != nullptr) { int total_replicas_num = partition->total_replica_num == 0 ? _num_replicas : partition->total_replica_num; - int load_required_replicas_num = partition->load_required_replica_num == 0 - ? (_num_replicas + 1) / 2 - : partition->load_required_replica_num; + int default_required_replicas_num = (_num_replicas + 1) / 2; + // Older FE createPartition responses omit these optional fields. Clamp the fallback to + // the selected destinations instead of rejecting a valid min_load_replica_num load. + int load_required_replicas_num = + partition->load_required_replica_num == 0 + ? std::min(default_required_replicas_num, selected_replicas_num) + : partition->load_required_replica_num; _tablet_replica_info.emplace( - tablet_id, std::make_pair(total_replicas_num, load_required_replicas_num)); + tablet_id, + TabletReplicaInfo {.total_replicas_num = total_replicas_num, + .load_required_replicas_num = load_required_replicas_num, + .selected_replicas_num = selected_replicas_num}); // Copy version gap backends info for this tablet if (auto it = partition->tablet_version_gap_backends.find(tablet_id); it != partition->tablet_version_gap_backends.end()) { _tablet_version_gap_backends[tablet_id] = it->second; } } else { - _tablet_replica_info.emplace(tablet_id, - std::make_pair(_num_replicas, (_num_replicas + 1) / 2)); + int load_required_replicas_num = std::min((_num_replicas + 1) / 2, selected_replicas_num); + _tablet_replica_info.emplace( + tablet_id, + TabletReplicaInfo {.total_replicas_num = _num_replicas, + .load_required_replicas_num = load_required_replicas_num, + .selected_replicas_num = selected_replicas_num}); + } + return Status::OK(); +} + +void VTabletWriter::_add_partition_involvement_commit_infos() { + if (config::is_cloud_mode()) { + return; + } + constexpr int64_t PARTITION_INVOLVEMENT_BACKEND_ID = -1; + std::vector commit_infos; + const auto& partition_ids = _tablet_finder->partition_ids(); + for (const auto* partition : _vpartition->get_partitions()) { + if (!partition_ids.contains(partition->id)) { + continue; + } + bool marker_added = false; + for (const auto& index : partition->indexes) { + for (int64_t tablet_id : index.tablets) { + if (tablet_id == V1_DUMMY_TABLET_ID) { + continue; + } + TTabletCommitInfo commit_info; + commit_info.__set_tabletId(tablet_id); + commit_info.__set_backendId(PARTITION_INVOLVEMENT_BACKEND_ID); + commit_infos.emplace_back(std::move(commit_info)); + marker_added = true; + break; + } + if (marker_added) { + break; + } + } + // A partition recorded by TabletFinder came from a routed row, so it must contain a real + // tablet. Tablet 0 belongs only to the synthetic empty-table bootstrap partition. + DORIS_CHECK(marker_added) << "no real tablet in touched partition " << partition->id; } + _state->add_tablet_commit_infos(commit_infos); } void VTabletWriter::_cancel_all_channel(Status status) { @@ -2032,6 +2141,7 @@ Status VTabletWriter::close(Status exec_status) { // If _close_status is not ok, all nodes have been canceled in try_close. if (_close_status.ok()) { + _add_partition_involvement_commit_infos(); auto status = Status::OK(); // BE id -> add_batch method counter std::unordered_map node_add_batch_counter_map; diff --git a/be/src/exec/sink/writer/vtablet_writer.h b/be/src/exec/sink/writer/vtablet_writer.h index 932e8b2b56c9b3..089e986c1e8efa 100644 --- a/be/src/exec/sink/writer/vtablet_writer.h +++ b/be/src/exec/sink/writer/vtablet_writer.h @@ -685,7 +685,9 @@ class VTabletWriter final : public AsyncResultWriter { void _do_try_close(RuntimeState* state, const Status& exec_status); - void _build_tablet_replica_info(const int64_t tablet_id, VOlapTablePartition* partition); + Status _build_tablet_replica_info(const int64_t tablet_id, VOlapTablePartition* partition); + + void _add_partition_involvement_commit_infos(); TDataSink _t_sink; @@ -789,8 +791,14 @@ class VTabletWriter final : public AsyncResultWriter { // reuse to avoid frequent memory allocation and release. std::vector _row_part_tablet_ids; - // tablet_id -> - std::unordered_map> _tablet_replica_info; + struct TabletReplicaInfo { + int total_replicas_num; + int load_required_replicas_num; + int selected_replicas_num; + }; + + // tablet_id -> configured and selected replica counts used by failure accounting + std::unordered_map _tablet_replica_info; // tablet_id -> set of backend_ids that have version gaps // these backends' success should not be counted for majority write diff --git a/be/src/exec/sink/writer/vtablet_writer_v2.cpp b/be/src/exec/sink/writer/vtablet_writer_v2.cpp index 13a38943d09e59..9f398a5701d9f2 100644 --- a/be/src/exec/sink/writer/vtablet_writer_v2.cpp +++ b/be/src/exec/sink/writer/vtablet_writer_v2.cpp @@ -32,6 +32,7 @@ #include #include +#include "common/check.h" #include "common/compiler_util.h" // IWYU pragma: keep #include "common/logging.h" #include "common/metrics/doris_metrics.h" @@ -115,7 +116,7 @@ Status VTabletWriterV2::_incremental_open_streams( VLOG_DEBUG << "incremental open stream (" << partition->id << ", " << tablet_id << ")"; } - _build_tablet_replica_info(tablet_id, partition); + RETURN_IF_ERROR(_build_tablet_replica_info(tablet_id, partition)); } } } @@ -357,31 +358,84 @@ Status VTabletWriterV2::_build_tablet_node_mapping() { _indexes_from_node[node].emplace_back(tablet); known_indexes.insert(index.index_id); } - _build_tablet_replica_info(tablet_id, partition); + RETURN_IF_ERROR(_build_tablet_replica_info(tablet_id, partition)); } } } return Status::OK(); } -void VTabletWriterV2::_build_tablet_replica_info(const int64_t tablet_id, - VOlapTablePartition* partition) { +Status VTabletWriterV2::_build_tablet_replica_info(const int64_t tablet_id, + VOlapTablePartition* partition) { + const auto* tablet_location = _location->find_tablet(tablet_id); + if (tablet_location == nullptr) { + return Status::InternalError("unknown tablet location, tablet_id={}", tablet_id); + } + int selected_replicas_num = cast_set(tablet_location->node_ids.size()); + if (selected_replicas_num == 0) { + return Status::InternalError("tablet {} has no selected replicas", tablet_id); + } + constexpr int64_t DUMMY_TABLET_ID = 0; + if (tablet_id == DUMMY_TABLET_ID) [[unlikely]] { + if (selected_replicas_num != 1) { + return Status::InternalError( + "bootstrap tablet {} must have exactly one selected replica, actual {}", + tablet_id, selected_replicas_num); + } + return _register_tablet_replica_info(tablet_id, partition == nullptr ? -1 : partition->id, + 1, 1, 1, {}); + } if (partition != nullptr) { int total_replicas_num = partition->total_replica_num == 0 ? _num_replicas : partition->total_replica_num; - int load_required_replicas_num = partition->load_required_replica_num == 0 - ? (_num_replicas + 1) / 2 - : partition->load_required_replica_num; - _tablet_replica_info[tablet_id] = - std::make_pair(total_replicas_num, load_required_replicas_num); - // Copy version gap backends info for this tablet + int default_required_replicas_num = (_num_replicas + 1) / 2; + // Older FE createPartition responses omit these optional fields. Clamp the fallback to + // the selected destinations instead of rejecting a valid min_load_replica_num load. + int load_required_replicas_num = + partition->load_required_replica_num == 0 + ? std::min(default_required_replicas_num, selected_replicas_num) + : partition->load_required_replica_num; + std::unordered_set selected_gap_backends; if (auto it = partition->tablet_version_gap_backends.find(tablet_id); it != partition->tablet_version_gap_backends.end()) { - _tablet_version_gap_backends[tablet_id] = it->second; + for (int64_t backend_id : it->second) { + if (std::ranges::find(tablet_location->node_ids, backend_id) != + tablet_location->node_ids.end()) { + selected_gap_backends.insert(backend_id); + } + } } - } else { - _tablet_replica_info[tablet_id] = std::make_pair(_num_replicas, (_num_replicas + 1) / 2); + return _register_tablet_replica_info(tablet_id, partition->id, total_replicas_num, + load_required_replicas_num, selected_replicas_num, + std::move(selected_gap_backends)); + } + return _register_tablet_replica_info(tablet_id, -1, _num_replicas, + std::min((_num_replicas + 1) / 2, selected_replicas_num), + selected_replicas_num, {}); +} + +Status VTabletWriterV2::_register_tablet_replica_info( + int64_t tablet_id, int64_t partition_id, int total_replicas_num, + int load_required_replicas_num, int selected_replicas_num, + std::unordered_set version_gap_backends) { + if (selected_replicas_num < load_required_replicas_num) { + return Status::InternalError( + "tablet {} in partition {} has {} selected replicas, fewer than required {}, " + "load_id={}", + tablet_id, partition_id, selected_replicas_num, load_required_replicas_num, + print_id(_load_id)); } + _tablet_replica_info[tablet_id] = + TabletReplicaInfo {.total_replicas_num = total_replicas_num, + .load_required_replicas_num = load_required_replicas_num, + .selected_replicas_num = selected_replicas_num}; + _tablet_version_gap_backends[tablet_id] = version_gap_backends; + return _load_stream_map->merge_tablet_replica_info( + tablet_id, LoadStreamMap::TabletReplicaInfo { + .total_replicas_num = total_replicas_num, + .load_required_replicas_num = load_required_replicas_num, + .selected_replicas_num = selected_replicas_num, + .version_gap_backends = std::move(version_gap_backends)}); } void VTabletWriterV2::_generate_rows_for_tablet(std::vector& row_part_tablet_ids, @@ -947,12 +1001,17 @@ bool VTabletWriterV2::_quorum_success( return true; } -int VTabletWriterV2::_load_required_replicas_num(int64_t tablet_id) { - auto [total_replicas_num, load_required_replicas_num] = _tablet_replica_info[tablet_id]; - if (total_replicas_num == 0) { +int VTabletWriterV2::_load_required_replicas_num(int64_t tablet_id) const { + LoadStreamMap::TabletReplicaInfo shared_replica_info; + if (_load_stream_map->get_tablet_replica_info(tablet_id, &shared_replica_info)) { + return shared_replica_info.load_required_replicas_num; + } + DORIS_CHECK(_tablet_replica_info.contains(tablet_id)); + const auto& replica_info = _tablet_replica_info.at(tablet_id); + if (replica_info.total_replicas_num == 0) { return (_num_replicas + 1) / 2; } - return load_required_replicas_num; + return replica_info.load_required_replicas_num; } int64_t VTabletWriterV2::_calc_max_wait_time_ms( @@ -1097,24 +1156,78 @@ Status VTabletWriterV2::_create_commit_info(std::vector& tabl }); for (auto& [tablet_id, failed_backends] : failed_tablets) { - // Version-gap replicas cannot contribute to quorum, even if this write succeeds. - // Count a backend only once when it also reported a write failure. - if (auto gap_it = _tablet_version_gap_backends.find(tablet_id); - gap_it != _tablet_version_gap_backends.end()) { - failed_backends.insert(gap_it->second.begin(), gap_it->second.end()); + RETURN_IF_ERROR(_check_failed_tablet(tablet_id, failed_backends, failed_reason[tablet_id], + load_stream_map)); + } + if (_state != nullptr && !failed_tablets.empty()) { + std::vector error_tablet_infos; + error_tablet_infos.reserve(failed_tablets.size()); + for (const auto& [tablet_id, _] : failed_tablets) { + TErrorTabletInfo error_info; + error_info.__set_tabletId(tablet_id); + error_info.__set_msg(failed_reason[tablet_id].to_string()); + error_tablet_infos.emplace_back(std::move(error_info)); } - auto [total_replicas_num, load_required_replicas_num] = _tablet_replica_info[tablet_id]; - int max_failed_replicas = total_replicas_num == 0 - ? (_num_replicas - 1) / 2 - : total_replicas_num - load_required_replicas_num; - if (std::cmp_greater(failed_backends.size(), max_failed_replicas)) { - LOG(INFO) << "tablet " << tablet_id - << " failed on majority backends (failed=" << failed_backends.size() - << ", max_failed=" << max_failed_replicas - << "): " << failed_reason[tablet_id]; - return Status::InternalError("tablet {} failed on majority backends: {}", tablet_id, - failed_reason[tablet_id]); + _state->add_error_tablet_infos(error_tablet_infos); + } + if (!config::is_cloud_mode()) { + // The final local writer may stop after quorum before a slow stream returns another + // writer's tablet result. tablets_to_commit is populated by every local writer before + // release(), so these markers keep every touched partition in FE's global quorum check. + load_stream_map->append_partition_involvement_commit_infos(tablet_commit_infos); + } + return Status::OK(); +} + +Status VTabletWriterV2::_check_failed_tablet( + int64_t tablet_id, std::unordered_set& failed_backends, + const Status& failed_reason, const std::shared_ptr& load_stream_map) const { + // Version-gap replicas cannot contribute to quorum, even if this write succeeds. + // Count a backend only once when it also reported a write failure. + if (auto gap_it = _tablet_version_gap_backends.find(tablet_id); + gap_it != _tablet_version_gap_backends.end()) { + failed_backends.insert(gap_it->second.begin(), gap_it->second.end()); + } + int configured_max_failed; + int selected_max_failed; + LoadStreamMap::TabletReplicaInfo shared_replica_info; + if (load_stream_map->get_tablet_replica_info(tablet_id, &shared_replica_info)) { + failed_backends.insert(shared_replica_info.version_gap_backends.begin(), + shared_replica_info.version_gap_backends.end()); + configured_max_failed = shared_replica_info.total_replicas_num - + shared_replica_info.load_required_replicas_num; + selected_max_failed = shared_replica_info.selected_replicas_num - + shared_replica_info.load_required_replicas_num; + } else { + auto replica_info_it = _tablet_replica_info.find(tablet_id); + if (replica_info_it == _tablet_replica_info.end()) { + // A receiver can return a dynamic tablet's final result to a different source BE, + // which never received that partition's replica policy. Shared-nothing FE combines + // its error marker with commit infos from every source and validates quorum. Cloud + // commit has no equivalent global error-tablet aggregation. + if (config::is_cloud_mode()) { + return Status::InternalError("missing replica information for failed tablet {}", + tablet_id); + } + return Status::OK(); } + const auto& replica_info = replica_info_it->second; + configured_max_failed = + replica_info.total_replicas_num == 0 + ? (_num_replicas - 1) / 2 + : replica_info.total_replicas_num - replica_info.load_required_replicas_num; + selected_max_failed = + replica_info.selected_replicas_num - replica_info.load_required_replicas_num; + } + DORIS_CHECK_GE(configured_max_failed, 0); + DORIS_CHECK_GE(selected_max_failed, 0); + int max_failed_replicas = std::min(configured_max_failed, selected_max_failed); + if (std::cmp_greater(failed_backends.size(), max_failed_replicas)) { + LOG(INFO) << "tablet " << tablet_id + << " failed on majority backends (failed=" << failed_backends.size() + << ", max_failed=" << max_failed_replicas << "): " << failed_reason; + return Status::InternalError("tablet {} failed on majority backends: {}", tablet_id, + failed_reason); } return Status::OK(); } diff --git a/be/src/exec/sink/writer/vtablet_writer_v2.h b/be/src/exec/sink/writer/vtablet_writer_v2.h index a3d86a613bbde1..8f137e8251bb50 100644 --- a/be/src/exec/sink/writer/vtablet_writer_v2.h +++ b/be/src/exec/sink/writer/vtablet_writer_v2.h @@ -117,6 +117,10 @@ class VTabletWriterV2 final : public AsyncResultWriter { Status _create_commit_info(std::vector& tablet_commit_infos, std::shared_ptr load_stream_map); + Status _check_failed_tablet(int64_t tablet_id, std::unordered_set& failed_backends, + const Status& failed_reason, + const std::shared_ptr& load_stream_map) const; + private: static constexpr int64_t CLOSE_WAIT_EVENT_FALLBACK_MS = 1000; @@ -130,7 +134,12 @@ class VTabletWriterV2 final : public AsyncResultWriter { Status _incremental_open_streams(const std::vector& partitions); - void _build_tablet_replica_info(const int64_t tablet_id, VOlapTablePartition* partition); + Status _build_tablet_replica_info(const int64_t tablet_id, VOlapTablePartition* partition); + + Status _register_tablet_replica_info(int64_t tablet_id, int64_t partition_id, + int total_replicas_num, int load_required_replicas_num, + int selected_replicas_num, + std::unordered_set version_gap_backends); Status _send_new_partition_batch(); @@ -157,7 +166,7 @@ class VTabletWriterV2 final : public AsyncResultWriter { const std::unordered_set>& unfinished_streams, const std::unordered_set& need_finish_tablets); - int _load_required_replicas_num(int64_t tablet_id); + int _load_required_replicas_num(int64_t tablet_id) const; Status _check_timeout(); @@ -253,8 +262,14 @@ class VTabletWriterV2 final : public AsyncResultWriter { // reuse to avoid frequent memory allocation and release. std::vector _row_part_tablet_ids; - // tablet_id -> - std::unordered_map> _tablet_replica_info; + struct TabletReplicaInfo { + int total_replicas_num; + int load_required_replicas_num; + int selected_replicas_num; + }; + + // tablet_id -> configured and selected replica counts used by failure accounting + std::unordered_map _tablet_replica_info; // tablet_id -> set of backend_ids that have version gaps std::unordered_map> _tablet_version_gap_backends; diff --git a/be/test/exec/sink/vtablet_writer_failure_test.cpp b/be/test/exec/sink/vtablet_writer_failure_test.cpp new file mode 100644 index 00000000000000..10ead6a48a9fe3 --- /dev/null +++ b/be/test/exec/sink/vtablet_writer_failure_test.cpp @@ -0,0 +1,192 @@ +// 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 "exec/sink/writer/vtablet_writer.h" +#include "testutil/mock/mock_runtime_state.h" + +namespace doris { +namespace { + +std::unique_ptr create_writer() { + TDataSink sink; + sink.__isset.olap_table_sink = true; + sink.olap_table_sink.num_replicas = 3; + auto writer = std::make_unique(sink, VExprContextSPtrs {}, nullptr, nullptr); + writer->_num_replicas = 3; + return writer; +} + +TEST(VTabletWriterFailureTest, FailureAllowanceUsesSelectedReplicas) { + auto writer = create_writer(); + IndexChannel channel(writer.get(), 10, nullptr); + struct TestCase { + int total; + int required; + int selected; + int max_failed; + }; + std::vector cases {{.total = 3, .required = 1, .selected = 1, .max_failed = 0}, + {.total = 3, .required = 1, .selected = 2, .max_failed = 1}, + {.total = 3, .required = 1, .selected = 3, .max_failed = 2}, + {.total = 3, .required = 2, .selected = 2, .max_failed = 0}, + {.total = 3, .required = 2, .selected = 3, .max_failed = 1}, + {.total = 5, .required = 2, .selected = 3, .max_failed = 1}}; + int64_t tablet_id = 100; + for (const auto& test_case : cases) { + writer->_tablet_replica_info.emplace( + tablet_id, + VTabletWriter::TabletReplicaInfo {.total_replicas_num = test_case.total, + .load_required_replicas_num = test_case.required, + .selected_replicas_num = test_case.selected}); + EXPECT_EQ(test_case.max_failed, channel._max_failed_replicas(tablet_id)); + ++tablet_id; + } +} + +TEST(VTabletWriterFailureTest, RepeatedFailureDoesNotConsumeAnotherDestination) { + auto writer = create_writer(); + writer->_tablet_replica_info.emplace( + 100, VTabletWriter::TabletReplicaInfo {.total_replicas_num = 3, + .load_required_replicas_num = 1, + .selected_replicas_num = 2}); + IndexChannel channel(writer.get(), 10, nullptr); + VNodeChannel node1(writer.get(), &channel, 1); + VNodeChannel node2(writer.get(), &channel, 2); + channel._tablets_by_channel[1].insert(100); + channel._tablets_by_channel[2].insert(100); + + channel.mark_as_failed(&node1, "first failure", 100); + EXPECT_TRUE(channel.check_intolerable_failure().ok()); + channel.mark_as_failed(&node1, "repeated failure", 100); + EXPECT_TRUE(channel.check_intolerable_failure().ok()); + channel.mark_as_failed(&node2, "last selected replica failed", 100); + EXPECT_FALSE(channel.check_intolerable_failure().ok()); +} + +TEST(VTabletWriterFailureTest, RejectsFewerSelectedReplicasThanRequired) { + auto writer = create_writer(); + TOlapTableLocationParam thrift_location; + TTabletLocation tablet_location; + tablet_location.tablet_id = 100; + tablet_location.node_ids = {1}; + thrift_location.tablets = {tablet_location}; + OlapTableLocationParam location(thrift_location); + writer->_location = &location; + writer->_tablet_replica_info.emplace( + 100, VTabletWriter::TabletReplicaInfo {.total_replicas_num = 3, + .load_required_replicas_num = 2, + .selected_replicas_num = 1}); + IndexChannel channel(writer.get(), 10, nullptr); + TTabletWithPartition tablet; + tablet.tablet_id = 100; + tablet.partition_id = 20; + + auto status = channel.init(nullptr, {tablet}); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("1 selected replicas, fewer than required 2"), + std::string::npos); +} + +TEST(VTabletWriterFailureTest, BootstrapTabletUsesSingleReplicaPolicy) { + auto writer = create_writer(); + TOlapTableLocationParam thrift_location; + TTabletLocation tablet_location; + tablet_location.tablet_id = 0; + tablet_location.node_ids = {1}; + thrift_location.tablets = {tablet_location}; + OlapTableLocationParam location(thrift_location); + writer->_location = &location; + + ASSERT_TRUE(writer->_build_tablet_replica_info(0, nullptr).ok()); + const auto& replica_info = writer->_tablet_replica_info.at(0); + EXPECT_EQ(1, replica_info.total_replicas_num); + EXPECT_EQ(1, replica_info.load_required_replicas_num); + EXPECT_EQ(1, replica_info.selected_replicas_num); + IndexChannel channel(writer.get(), 10, nullptr); + EXPECT_EQ(0, channel._max_failed_replicas(0)); +} + +TEST(VTabletWriterFailureTest, NonlocalTabletErrorIsReportedForGlobalValidation) { + auto writer = create_writer(); + writer->_tablet_replica_info.emplace( + 200, VTabletWriter::TabletReplicaInfo {.total_replicas_num = 3, + .load_required_replicas_num = 2, + .selected_replicas_num = 3}); + IndexChannel channel(writer.get(), 10, nullptr); + VNodeChannel node(writer.get(), &channel, 1); + // This writer opened tablet 200, but the shared receiver returns tablet 100's late error. + channel._tablets_by_channel[1].insert(200); + channel.mark_as_failed(&node, "late error from another sender", 100); + + EXPECT_TRUE(channel.check_intolerable_failure().ok()); + MockRuntimeState state; + channel.set_error_tablet_in_state(&state); + auto error_infos = state.error_tablet_infos(); + ASSERT_EQ(1, error_infos.size()); + EXPECT_EQ(100, error_infos[0].tabletId); +} + +TEST(VTabletWriterFailureTest, BootstrapOnlySenderPreservesRealTabletError) { + auto writer = create_writer(); + IndexChannel channel(writer.get(), 10, nullptr); + VNodeChannel node(writer.get(), &channel, 1); + // Tablet 0 is excluded from backend bookkeeping, so this sender has no entry for BE1. + channel.mark_as_failed(&node, "late error returned to bootstrap sender", 100); + + EXPECT_TRUE(channel.check_intolerable_failure().ok()); + MockRuntimeState state; + channel.set_error_tablet_in_state(&state); + auto error_infos = state.error_tablet_infos(); + ASSERT_EQ(1, error_infos.size()); + EXPECT_EQ(100, error_infos[0].tabletId); +} + +TEST(VTabletWriterFailureTest, BootstrapOnlySenderRejectsOverallChannelFailure) { + auto writer = create_writer(); + IndexChannel channel(writer.get(), 10, nullptr); + VNodeChannel node(writer.get(), &channel, 1); + channel.mark_as_failed(&node, "whole rpc failed", -1); + + auto status = channel.check_intolerable_failure(); + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("no local real tablet"), std::string::npos); +} + +TEST(VTabletWriterFailureTest, OmittedReplicaPolicyUsesSelectedDestinations) { + auto writer = create_writer(); + TOlapTableLocationParam thrift_location; + TTabletLocation tablet_location; + tablet_location.tablet_id = 100; + tablet_location.node_ids = {1}; + thrift_location.tablets = {tablet_location}; + OlapTableLocationParam location(thrift_location); + writer->_location = &location; + ASSERT_TRUE(writer->_build_tablet_replica_info(100, nullptr).ok()); + const auto& replica_info = writer->_tablet_replica_info.at(100); + EXPECT_EQ(3, replica_info.total_replicas_num); + EXPECT_EQ(1, replica_info.load_required_replicas_num); + EXPECT_EQ(1, replica_info.selected_replicas_num); +} + +} // namespace +} // namespace doris diff --git a/be/test/exec/sink/vtablet_writer_v2_test.cpp b/be/test/exec/sink/vtablet_writer_v2_test.cpp index c2d43c3995a24f..d23a2381ee4111 100644 --- a/be/test/exec/sink/vtablet_writer_v2_test.cpp +++ b/be/test/exec/sink/vtablet_writer_v2_test.cpp @@ -66,7 +66,9 @@ static void add_stream(std::shared_ptr load_stream_map, int64_t n } } -static std::unique_ptr create_vtablet_writer(int num_replicas = 3) { +static std::unique_ptr create_vtablet_writer(int num_replicas = 3, + int required_replicas = -1, + int selected_replicas = -1) { TDataSink t_sink; t_sink.__isset.olap_table_sink = true; t_sink.olap_table_sink.num_replicas = num_replicas; @@ -74,10 +76,20 @@ static std::unique_ptr create_vtablet_writer(int num_replicas = std::shared_ptr dep = nullptr; std::shared_ptr fin_dep = nullptr; auto writer = std::make_unique(t_sink, output_exprs, dep, fin_dep); + writer->_num_replicas = num_replicas; - int required_replicas = num_replicas / 2 + 1; - writer->_tablet_replica_info[1] = std::make_pair(num_replicas, required_replicas); - writer->_tablet_replica_info[2] = std::make_pair(num_replicas, required_replicas); + if (required_replicas < 0) { + required_replicas = num_replicas / 2 + 1; + } + if (selected_replicas < 0) { + selected_replicas = num_replicas; + } + for (int64_t tablet_id : {1, 2}) { + writer->_tablet_replica_info[tablet_id] = + VTabletWriterV2::TabletReplicaInfo {.total_replicas_num = num_replicas, + .load_required_replicas_num = required_replicas, + .selected_replicas_num = selected_replicas}; + } return writer; } @@ -414,6 +426,176 @@ TEST_F(TestVTabletWriterV2, fail_one) { ASSERT_EQ(tablet_commit_infos.size(), 5); } +TEST_F(TestVTabletWriterV2, selectedReplicaLateFailureRejectsMixedPartitions) { + UniqueId load_id; + auto load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + add_stream(load_stream_map, 1001, {}, + {{1, Status::InternalError("late close failed")}, + {2, Status::InternalError("late close failed")}}); + add_stream(load_stream_map, 1004, {2}, {}); + add_stream(load_stream_map, 1005, {2}, {}); + + auto writer = create_vtablet_writer(3, 1, 1); + writer->_tablet_replica_info[2].selected_replicas_num = 3; + std::vector commit_infos; + auto status = writer->_create_commit_info(commit_infos, load_stream_map); + + EXPECT_FALSE(status.ok()); + EXPECT_NE(status.to_string().find("late close failed"), std::string::npos); + ASSERT_EQ(2, commit_infos.size()); + EXPECT_TRUE( + std::ranges::all_of(commit_infos, [](const auto& info) { return info.tabletId == 2; })); +} + +TEST_F(TestVTabletWriterV2, finalWriterUsesSharedIncrementalReplicaPolicy) { + UniqueId load_id; + auto load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + add_stream(load_stream_map, 1001, {}, {{1, Status::InternalError("late close failed")}}); + add_stream(load_stream_map, 1002, {1}, {}); + add_stream(load_stream_map, 1003, {1}, {}); + ASSERT_TRUE( + load_stream_map + ->merge_tablet_replica_info( + 1, LoadStreamMap::TabletReplicaInfo {.total_replicas_num = 3, + .load_required_replicas_num = 2, + .selected_replicas_num = 3, + .version_gap_backends = {}}) + .ok()); + + auto final_writer = create_vtablet_writer(); + final_writer->_tablet_replica_info.erase(1); + std::vector commit_infos; + auto status = final_writer->_create_commit_info(commit_infos, load_stream_map); + + EXPECT_TRUE(status.ok()) << status; + ASSERT_EQ(2, commit_infos.size()); +} + +TEST_F(TestVTabletWriterV2, sharedReplicaPolicyKeepsStrictestRefreshedRequirement) { + UniqueId load_id; + auto load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + auto policy = [](int required) { + return LoadStreamMap::TabletReplicaInfo {.total_replicas_num = 3, + .load_required_replicas_num = required, + .selected_replicas_num = 3, + .version_gap_backends = {}}; + }; + ASSERT_TRUE(load_stream_map->merge_tablet_replica_info(1, policy(2)).ok()); + ASSERT_TRUE(load_stream_map->merge_tablet_replica_info(1, policy(1)).ok()); + LoadStreamMap::TabletReplicaInfo merged; + ASSERT_TRUE(load_stream_map->get_tablet_replica_info(1, &merged)); + EXPECT_EQ(2, merged.load_required_replicas_num); + + auto reverse_map = std::make_shared(load_id, src_id + 1, 1, 1, nullptr); + ASSERT_TRUE(reverse_map->merge_tablet_replica_info(1, policy(1)).ok()); + ASSERT_TRUE(reverse_map->merge_tablet_replica_info(1, policy(2)).ok()); + ASSERT_TRUE(reverse_map->get_tablet_replica_info(1, &merged)); + EXPECT_EQ(2, merged.load_required_replicas_num); + + auto writer = create_vtablet_writer(3, 1, 3); + writer->_load_stream_map = reverse_map; + EXPECT_EQ(2, writer->_load_required_replicas_num(1)); +} + +TEST_F(TestVTabletWriterV2, sharedReplicaPolicyKeepsStrictestTotalRefreshAllowance) { + UniqueId load_id; + auto total_refresh = std::make_shared(load_id, src_id, 1, 1, nullptr); + LoadStreamMap::TabletReplicaInfo policy {.total_replicas_num = 3, + .load_required_replicas_num = 2, + .selected_replicas_num = 3, + .version_gap_backends = {}}; + auto refreshed_total = policy; + refreshed_total.total_replicas_num = 4; + ASSERT_TRUE(total_refresh->merge_tablet_replica_info(1, policy).ok()); + ASSERT_TRUE(total_refresh->merge_tablet_replica_info(1, refreshed_total).ok()); + LoadStreamMap::TabletReplicaInfo merged; + ASSERT_TRUE(total_refresh->get_tablet_replica_info(1, &merged)); + // The stricter configured allowance is min(3-2, 4-2)=1. + EXPECT_EQ(3, merged.total_replicas_num); + EXPECT_EQ(2, merged.load_required_replicas_num); +} + +TEST_F(TestVTabletWriterV2, toleratedLateFailureReportsTabletParticipation) { + UniqueId load_id; + auto load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + add_stream(load_stream_map, 1001, {}, {{1, Status::InternalError("late close failed")}}); + add_stream(load_stream_map, 1002, {1}, {}); + add_stream(load_stream_map, 1003, {1}, {}); + + auto writer = create_vtablet_writer(); + MockRuntimeState state; + writer->_state = &state; + std::vector commit_infos; + ASSERT_TRUE(writer->_create_commit_info(commit_infos, load_stream_map).ok()); + + auto error_infos = state.error_tablet_infos(); + ASSERT_EQ(1, error_infos.size()); + EXPECT_EQ(1, error_infos[0].tabletId); + EXPECT_NE(error_infos[0].msg.find("late close failed"), std::string::npos); +} + +TEST_F(TestVTabletWriterV2, nonlocalDynamicTabletFailureIsReportedForGlobalValidation) { + UniqueId load_id; + auto load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + add_stream(load_stream_map, 1001, {}, {{100, Status::InternalError("nonlocal failure")}}); + + auto writer = create_vtablet_writer(); + MockRuntimeState state; + writer->_state = &state; + std::vector commit_infos; + ASSERT_TRUE(writer->_create_commit_info(commit_infos, load_stream_map).ok()); + + auto error_infos = state.error_tablet_infos(); + ASSERT_EQ(1, error_infos.size()); + EXPECT_EQ(100, error_infos[0].tabletId); +} + +TEST_F(TestVTabletWriterV2, missingStreamResultStillReportsPartitionParticipation) { + UniqueId load_id; + auto load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + PTabletID first_tablet; + first_tablet.set_tablet_id(100); + first_tablet.set_partition_id(10); + PTabletID second_tablet; + second_tablet.set_tablet_id(200); + second_tablet.set_partition_id(20); + load_stream_map->save_tablets_to_commit(1001, {first_tablet, second_tablet}); + + auto writer = create_vtablet_writer(); + std::vector commit_infos; + ASSERT_TRUE(writer->_create_commit_info(commit_infos, load_stream_map).ok()); + + ASSERT_EQ(2, commit_infos.size()); + EXPECT_TRUE(std::ranges::all_of(commit_infos, + [](const auto& info) { return info.backendId == -1; })); + std::unordered_set involved_tablets; + for (const auto& info : commit_infos) { + involved_tablets.insert(info.tabletId); + } + EXPECT_EQ((std::unordered_set {100, 200}), involved_tablets); +} + +TEST_F(TestVTabletWriterV2, omittedReplicaPolicyUsesSelectedDestinations) { + auto writer = create_vtablet_writer(); + TOlapTableLocationParam thrift_location; + TTabletLocation tablet_location; + tablet_location.tablet_id = 100; + tablet_location.node_ids = {1}; + thrift_location.tablets = {tablet_location}; + OlapTableLocationParam location(thrift_location); + writer->_location = &location; + UniqueId load_id; + writer->_load_stream_map = std::make_shared(load_id, src_id, 1, 1, nullptr); + ASSERT_TRUE(writer->_build_tablet_replica_info(100, nullptr).ok()); + const auto& local_info = writer->_tablet_replica_info.at(100); + EXPECT_EQ(3, local_info.total_replicas_num); + EXPECT_EQ(1, local_info.load_required_replicas_num); + EXPECT_EQ(1, local_info.selected_replicas_num); + LoadStreamMap::TabletReplicaInfo shared_info; + ASSERT_TRUE(writer->_load_stream_map->get_tablet_replica_info(100, &shared_info)); + EXPECT_EQ(1, shared_info.load_required_replicas_num); +} + TEST_F(TestVTabletWriterV2, commit_info_with_results_split_across_sources) { UniqueId load_id; auto first_source = std::make_shared(load_id, src_id, 1, 1, nullptr); diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java index c7fdb68ec72a8c..542096c7a95dba 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java +++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java @@ -240,12 +240,19 @@ protected Multimap getNormalReplicaBackendPathMapImpl(String beEndpo } // return map of (BE id -> path hash) of normal replicas - // for load plan. public Multimap getNormalReplicaBackendPathMap() throws UserException { TabletSlidingWindowAccessStats.recordTablet(getId()); return getNormalReplicaBackendPathMapImpl(null, (rep, be) -> rep.getBackendId()); } + // return map of (BE id -> path hash) of replicas eligible for a new load plan + public Multimap getLoadReplicaBackendPathMap() throws UserException { + Multimap pathMap = getNormalReplicaBackendPathMap(); + SystemInfoService infoService = Env.getCurrentSystemInfo(); + pathMap.entries().removeIf(entry -> !infoService.checkBackendLoadAvailable(entry.getKey())); + return pathMap; + } + // When a BE reports a missing version, lastFailedVersion is set. When a write fails on a replica, // lastFailedVersion is set. // for query diff --git a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java index 541570b079e77a..d3a9003419a5e3 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java +++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudTablet.java @@ -98,6 +98,21 @@ public Multimap getNormalReplicaBackendPathMap(String beEndpoint) th return backendPathMapReprocess(pathMap); } + @Override + public Multimap getLoadReplicaBackendPathMap() throws UserException { + return getNormalReplicaBackendPathMap(); + } + + public Multimap getLoadReplicaBackendPathMapByClusterId(String clusterId) throws UserException { + // The logical replica has already resolved to one physical BE. Post-filtering that single + // result cannot reroute a load and may cause a false REPLICA_FEW_ERR during graceful shutdown. + return getNormalReplicaBackendPathMapByClusterId(clusterId); + } + + public Multimap getLoadReplicaBackendPathMap(String beEndpoint) throws UserException { + return getNormalReplicaBackendPathMap(beEndpoint); + } + @Override public void addReplica(Replica replica, boolean isRestore) { this.replica = replica; diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java index 154beaac8b5e11..1da9734bf2a52f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java +++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java @@ -1204,8 +1204,8 @@ public TOlapTableLocationParam createDummyLocation(OlapTable table) throws UserE public static Multimap getChangelogColocatedReplicaBackendPathMap( Tablet baseTablet, Tablet changelogTablet) throws UserException { - Multimap baseBePathsMap = baseTablet.getNormalReplicaBackendPathMap(); - Multimap changelogBePathsMap = changelogTablet.getNormalReplicaBackendPathMap(); + Multimap baseBePathsMap = baseTablet.getLoadReplicaBackendPathMap(); + Multimap changelogBePathsMap = changelogTablet.getLoadReplicaBackendPathMap(); changelogBePathsMap.entries().removeIf( entry -> !baseBePathsMap.containsEntry(entry.getKey(), entry.getValue())); return changelogBePathsMap; @@ -1237,7 +1237,7 @@ private TOlapTableLocationParam createLocation(long dbId, OlapTable table) throw .getCurrentClusterId(); } bePathsMap = ((CloudTablet) tablet) - .getNormalReplicaBackendPathMapByClusterId(cachedClusterId); + .getLoadReplicaBackendPathMapByClusterId(cachedClusterId); } else if (index.isChangelog()) { long baseTabletId = tablet.getChangelogBaseTabletId(); Tablet baseTablet = partition.getBaseIndex().getTablet(baseTabletId); @@ -1246,13 +1246,13 @@ private TOlapTableLocationParam createLocation(long dbId, OlapTable table) throw tablet.getId(), baseTabletId, partition.getId()); bePathsMap = getChangelogColocatedReplicaBackendPathMap(baseTablet, tablet); } else { - bePathsMap = tablet.getNormalReplicaBackendPathMap(); + bePathsMap = tablet.getLoadReplicaBackendPathMap(); } if (bePathsMap.keySet().size() < loadRequiredReplicaNum) { errMsgBuilder.append("tablet ").append(tablet.getId()) - .append(" alive replica num ").append(bePathsMap.keySet().size()) + .append(" available replica num ").append(bePathsMap.keySet().size()) .append(" < load required replica num ").append(loadRequiredReplicaNum) - .append(", alive backends: [") + .append(", available backends: [") .append(StringUtils.join(bePathsMap.keySet(), ",")) .append("]"); if (!Config.isCloudMode()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java index 99521b61ad0921..dbab267bba06f2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java @@ -125,6 +125,7 @@ import org.apache.doris.thrift.TTopnFilterDesc; import org.apache.doris.thrift.TUniqueId; import org.apache.doris.transaction.CommitDataSerializer; +import org.apache.doris.transaction.ErrorTabletInfo; import org.apache.doris.transaction.Transaction; import com.google.common.annotations.VisibleForTesting; @@ -1244,6 +1245,9 @@ private void updateCommitInfos(List commitInfos) { } private void updateErrorTabletInfos(List errorTabletInfos) { + if (!Config.isCloudMode()) { + updateCommitInfos(ErrorTabletInfo.toInvolvedTabletCommitInfos(errorTabletInfos)); + } lock.lock(); try { if (this.errorTabletInfos.size() <= Config.max_error_tablet_of_broker_load) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/LoadContext.java b/fe/fe-core/src/main/java/org/apache/doris/qe/LoadContext.java index 6fe143e3786996..5027a001e41ff2 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/qe/LoadContext.java +++ b/fe/fe-core/src/main/java/org/apache/doris/qe/LoadContext.java @@ -24,6 +24,7 @@ import org.apache.doris.task.LoadEtlTask; import org.apache.doris.thrift.TErrorTabletInfo; import org.apache.doris.thrift.TTabletCommitInfo; +import org.apache.doris.transaction.ErrorTabletInfo; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; @@ -151,6 +152,9 @@ public List getExportFiles() { } public synchronized void updateErrorTabletInfos(List errorTabletInfos) { + if (!Config.isCloudMode()) { + updateCommitInfos(ErrorTabletInfo.toInvolvedTabletCommitInfos(errorTabletInfos)); + } if (this.errorTabletInfos.size() <= Config.max_error_tablet_of_broker_load) { this.errorTabletInfos.addAll(errorTabletInfos.stream().limit(Config.max_error_tablet_of_broker_load - this.errorTabletInfos.size()).collect(Collectors.toList())); diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java index 50873fb01d6da3..ef55e547b26ffe 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java +++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java @@ -4917,20 +4917,21 @@ public TCreatePartitionResult createPartition(TCreatePartitionRequest request) t if (tablet instanceof CloudTablet) { CloudTablet cloudTablet = (CloudTablet) tablet; if (hasBeEndpoint) { - bePathsMap = cloudTablet.getNormalReplicaBackendPathMap(request.be_endpoint); + bePathsMap = cloudTablet.getLoadReplicaBackendPathMap(request.be_endpoint); } else { if (cachedClusterId == null) { cachedClusterId = ((CloudSystemInfoService) Env.getCurrentSystemInfo()) .getCurrentClusterId(); } - bePathsMap = cloudTablet.getNormalReplicaBackendPathMapByClusterId(cachedClusterId); + bePathsMap = cloudTablet.getLoadReplicaBackendPathMapByClusterId(cachedClusterId); } } else if (tabletSnapshot.isChangelog()) { bePathsMap = OlapTableSink.getChangelogColocatedReplicaBackendPathMap( tabletSnapshot.changelogBaseTablet, tablet); } else { - bePathsMap = tablet.getNormalReplicaBackendPathMap(); + bePathsMap = tablet.getLoadReplicaBackendPathMap(); } + checkLoadBackendPathNotEmpty(tableId, partitionId, tablet.getId(), bePathsMap); } catch (UserException ex) { errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage())); result.setStatus(errorStatus); @@ -5250,21 +5251,22 @@ public TReplacePartitionResult replacePartition(TReplacePartitionRequest request if (tablet instanceof CloudTablet) { CloudTablet cloudTablet = (CloudTablet) tablet; if (replaceHasBeEndpoint) { - bePathsMap = cloudTablet.getNormalReplicaBackendPathMap(request.be_endpoint); + bePathsMap = cloudTablet.getLoadReplicaBackendPathMap(request.be_endpoint); } else { if (replaceCachedClusterId == null) { replaceCachedClusterId = ((CloudSystemInfoService) Env.getCurrentSystemInfo()) .getCurrentClusterId(); } bePathsMap = cloudTablet - .getNormalReplicaBackendPathMapByClusterId(replaceCachedClusterId); + .getLoadReplicaBackendPathMapByClusterId(replaceCachedClusterId); } } else if (tabletSnapshot.isChangelog()) { bePathsMap = OlapTableSink.getChangelogColocatedReplicaBackendPathMap( tabletSnapshot.changelogBaseTablet, tablet); } else { - bePathsMap = tablet.getNormalReplicaBackendPathMap(); + bePathsMap = tablet.getLoadReplicaBackendPathMap(); } + checkLoadBackendPathNotEmpty(tableId, partitionId, tablet.getId(), bePathsMap); } catch (UserException ex) { errorStatus.setErrorMsgs(Lists.newArrayList(ex.getMessage())); result.setStatus(errorStatus); @@ -5381,6 +5383,17 @@ private TTabletLocation createLocation(List nodeIds) { } } + private static void checkLoadBackendPathNotEmpty(long tableId, long partitionId, long tabletId, + Multimap backendPathMap) throws UserException { + // The legacy writer accepts an empty location and can omit the entire partition from commit infos. + // Reject it before the location is cached or returned so the load fails instead of losing rows silently. + if (backendPathMap.isEmpty()) { + throw new UserException(InternalErrorCode.REPLICA_FEW_ERR, + String.format("no load available backend for tablet %d in partition %d of table %d", + tabletId, partitionId, tableId)); + } + } + private static List snapshotPartitionResultsByName(OlapTable olapTable, Collection partitionNames, boolean loadToSingleTablet, boolean enableAdaptiveRandomBucket, String resultName) throws UserException { @@ -5394,7 +5407,7 @@ private static List snapshotPartitionResultsByName(Olap "partition %s was dropped concurrently while building %s result, please retry", partitionName, resultName)); } - partitionSnapshots.add(snapshotPartitionResult(partitionInfo, partColNum, partition, partitionName, + partitionSnapshots.add(snapshotPartitionResult(olapTable, partColNum, partition, partitionName, loadToSingleTablet, enableAdaptiveRandomBucket, resultName)); } return partitionSnapshots; @@ -5413,16 +5426,17 @@ private static List snapshotPartitionResultsById(OlapTa "partition %d was dropped concurrently while building %s result, please retry", partitionId, resultName)); } - partitionSnapshots.add(snapshotPartitionResult(partitionInfo, partColNum, partition, + partitionSnapshots.add(snapshotPartitionResult(olapTable, partColNum, partition, String.valueOf(partitionId), loadToSingleTablet, enableAdaptiveRandomBucket, resultName)); } return partitionSnapshots; } - private static PartitionResultSnapshot snapshotPartitionResult(PartitionInfo partitionInfo, int partColNum, + private static PartitionResultSnapshot snapshotPartitionResult(OlapTable olapTable, int partColNum, Partition partition, String partitionLabel, boolean loadToSingleTablet, boolean enableAdaptiveRandomBucket, String resultName) throws UserException { long partitionId = partition.getId(); + PartitionInfo partitionInfo = olapTable.getPartitionInfo(); PartitionItem partitionItem = partitionInfo.getItem(partitionId); if (partitionItem == null) { throw new UserException(String.format( @@ -5458,7 +5472,9 @@ private static PartitionResultSnapshot snapshotPartitionResult(PartitionInfo par boolean randomDistribution = partition.getDistributionInfo().getType() == DistributionInfo.DistributionInfoType.RANDOM; boolean cacheLoadTabletIdx = (loadToSingleTablet || enableAdaptiveRandomBucket) && randomDistribution; - int quorum = partitionInfo.getReplicaAllocation(partitionId).getTotalReplicaNum() / 2 + 1; + int quorum = olapTable.getLoadRequiredReplicaNum(partitionId); + tPartition.setTotalReplicaNum(olapTable.getPartitionTotalReplicasNum(partitionId)); + tPartition.setLoadRequiredReplicaNum(quorum); return new PartitionResultSnapshot(partition, partitionId, tPartition, partitionTabletSnapshot, quorum, cacheLoadTabletIdx); } diff --git a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java index 5713be5b965395..fdc59d052627d1 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java +++ b/fe/fe-core/src/main/java/org/apache/doris/system/SystemInfoService.java @@ -358,6 +358,14 @@ public boolean checkBackendScheduleAvailable(long backendId) { return true; } + public boolean checkBackendLoadAvailable(long backendId) { + Backend backend = getAllClusterBackendsNoException().get(backendId); + if (backend == null || !backend.isLoadAvailable()) { + return false; + } + return true; + } + public boolean checkBackendAlive(long backendId) { Backend backend = getAllClusterBackendsNoException().get(backendId); if (backend == null || !backend.isAlive()) { diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java index 2d37d7b0bde912..30b37ae0f99595 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java +++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/DatabaseTransactionMgr.java @@ -579,7 +579,12 @@ private void checkCommitStatus(List tableList, TransactionState transacti if (!tabletToBackends.containsKey(tabletId)) { tabletToBackends.put(tabletId, new HashSet<>()); } - tabletToBackends.get(tabletId).add(tabletCommitInfos.get(i).getBackendId()); + long backendId = tabletCommitInfos.get(i).getBackendId(); + // Partition involvement markers use a negative backend only to include the + // partition in quorum validation. They must never count as successful replicas. + if (backendId >= 0) { + tabletToBackends.get(tabletId).add(backendId); + } } List tabletSuccReplicas = Lists.newArrayList(); List tabletWriteFailedReplicas = Lists.newArrayList(); diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/ErrorTabletInfo.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/ErrorTabletInfo.java index 64108884b83b8e..9473f06a87a550 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/transaction/ErrorTabletInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/ErrorTabletInfo.java @@ -18,6 +18,7 @@ package org.apache.doris.transaction; import org.apache.doris.thrift.TErrorTabletInfo; +import org.apache.doris.thrift.TTabletCommitInfo; import com.google.common.collect.Lists; @@ -47,4 +48,16 @@ public static List fromThrift(List errorTable } return errorInfos; } + + public static List toInvolvedTabletCommitInfos( + List errorTabletInfos) { + List commitInfos = Lists.newArrayList(); + for (TErrorTabletInfo errorInfo : errorTabletInfos) { + if (errorInfo.isSetTabletId()) { + commitInfos.add(new TTabletCommitInfo( + errorInfo.getTabletId(), TabletCommitInfo.PARTITION_INVOLVEMENT_BACKEND_ID)); + } + } + return commitInfos; + } } diff --git a/fe/fe-core/src/main/java/org/apache/doris/transaction/TabletCommitInfo.java b/fe/fe-core/src/main/java/org/apache/doris/transaction/TabletCommitInfo.java index 577fd657934996..19d5d14faa1f9f 100644 --- a/fe/fe-core/src/main/java/org/apache/doris/transaction/TabletCommitInfo.java +++ b/fe/fe-core/src/main/java/org/apache/doris/transaction/TabletCommitInfo.java @@ -25,6 +25,7 @@ import java.util.List; public class TabletCommitInfo { + public static final long PARTITION_INVOLVEMENT_BACKEND_ID = -1L; private long tabletId; private long backendId; diff --git a/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletTest.java b/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletTest.java index 8b70af2a040398..a6df773e29f263 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/catalog/TabletTest.java @@ -27,6 +27,7 @@ import org.apache.doris.system.SystemInfoService; import org.apache.doris.thrift.TStorageMedium; +import com.google.common.collect.Multimap; import com.google.common.collect.Sets; import com.google.gson.JsonObject; import com.google.gson.JsonParser; @@ -109,6 +110,17 @@ public void getMethodTest() { Assertions.assertEquals("tabletId=" + newTabletId, tablet.toString()); } + @Test + public void testMetadataReplicaSelectionKeepsLoadDisabledBackendDuringDecommission() throws Exception { + infoService.getBackend(2L).setLoadDisabled(true); + replica3.updateLastFailedVersion(-1L); + replica3.setState(ReplicaState.DECOMMISSION); + + Assertions.assertEquals(Sets.newHashSet(1L, 2L, 3L), Sets.newHashSet(tablet.getNormalReplicaBackendIds())); + Multimap loadBackendPaths = tablet.getLoadReplicaBackendPathMap(); + Assertions.assertEquals(Sets.newHashSet(1L, 3L), loadBackendPaths.keySet()); + } + @Test public void testGetReplicaStatsOnlyUsesNormalReplicas() { Tablet statsTablet = new LocalTablet(2); diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/OlapTableSinkTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/OlapTableSinkTest.java index 7713a687fa9c6a..a7c1f43207fca7 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/planner/OlapTableSinkTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/planner/OlapTableSinkTest.java @@ -18,7 +18,16 @@ package org.apache.doris.planner; import org.apache.doris.catalog.Env; +import org.apache.doris.catalog.MaterializedIndex; +import org.apache.doris.catalog.MaterializedIndex.IndexExtState; import org.apache.doris.catalog.OlapTable; +import org.apache.doris.catalog.Partition; +import org.apache.doris.catalog.PartitionInfo; +import org.apache.doris.catalog.Tablet; +import org.apache.doris.cloud.catalog.CloudTablet; +import org.apache.doris.cloud.system.CloudSystemInfoService; +import org.apache.doris.common.Config; +import org.apache.doris.common.Status; import org.apache.doris.planner.OlapTableSink.AdaptiveBucketAssignment; import org.apache.doris.planner.OlapTableSink.AdaptiveIndexBucketAssignment; import org.apache.doris.system.Backend; @@ -26,9 +35,13 @@ import org.apache.doris.thrift.TOlapTableIndexTablets; import org.apache.doris.thrift.TOlapTableLocationParam; import org.apache.doris.thrift.TOlapTablePartition; +import org.apache.doris.thrift.TOlapTableSink; import org.apache.doris.thrift.TTabletLocation; +import com.google.common.collect.HashMultimap; import com.google.common.collect.ImmutableMap; +import com.google.common.collect.Multimap; +import com.google.common.collect.Sets; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.MockedStatic; @@ -40,6 +53,92 @@ import java.util.Map; public class OlapTableSinkTest { + @Test + public void testCreateLocationFiltersLoadUnavailableBackendsInSharedNothingMode() throws Exception { + String originalDeployMode = Config.deploy_mode; + String originalCloudUniqueId = Config.cloud_unique_id; + SystemInfoService systemInfoService = Mockito.mock(SystemInfoService.class); + Mockito.when(systemInfoService.checkBackendLoadAvailable(1L)).thenReturn(true); + Mockito.when(systemInfoService.checkBackendLoadAvailable(2L)).thenReturn(false); + Mockito.when(systemInfoService.checkBackendLoadAvailable(3L)).thenReturn(true); + Mockito.when(systemInfoService.checkExceedDiskCapacityLimit(Mockito.any(), Mockito.eq(true))) + .thenReturn(Status.OK); + + Tablet tablet = Mockito.mock(Tablet.class, Mockito.CALLS_REAL_METHODS); + Multimap backendPaths = HashMultimap.create(); + backendPaths.put(1L, 101L); + backendPaths.put(2L, 102L); + backendPaths.put(3L, 103L); + Mockito.doReturn(backendPaths).when(tablet).getNormalReplicaBackendPathMap(); + OlapTable table = mockTableWithSingleTablet(tablet, 2); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService); + Config.deploy_mode = ""; + Config.cloud_unique_id = ""; + + TOlapTableSink thriftSink = new TOlapTableSink(); + thriftSink.setDbId(100L); + OlapTableSink sink = new OlapTableSink(table, null, Collections.singletonList(10L)); + TOlapTableLocationParam location = sink.initLocationParam(thriftSink); + + Assertions.assertEquals(Sets.newHashSet(1L, 3L), + Sets.newHashSet(location.getTablets().get(0).getNodeIds())); + } finally { + Config.deploy_mode = originalDeployMode; + Config.cloud_unique_id = originalCloudUniqueId; + } + } + + @Test + public void testCreateLocationPreservesResolvedBackendInCloudMode() throws Exception { + String originalDeployMode = Config.deploy_mode; + String originalCloudUniqueId = Config.cloud_unique_id; + CloudSystemInfoService systemInfoService = Mockito.mock(CloudSystemInfoService.class); + Mockito.when(systemInfoService.getCurrentClusterId()).thenReturn("cluster-id"); + Mockito.when(systemInfoService.checkExceedDiskCapacityLimit(Mockito.any(), Mockito.eq(true))) + .thenReturn(Status.OK); + + CloudTablet tablet = Mockito.mock(CloudTablet.class, Mockito.CALLS_REAL_METHODS); + Multimap backendPaths = HashMultimap.create(); + backendPaths.put(2L, 102L); + Mockito.doReturn(backendPaths).when(tablet).getNormalReplicaBackendPathMapByClusterId("cluster-id"); + OlapTable table = mockTableWithSingleTablet(tablet, 1); + + try (MockedStatic mockedEnv = Mockito.mockStatic(Env.class)) { + mockedEnv.when(Env::getCurrentSystemInfo).thenReturn(systemInfoService); + Config.deploy_mode = "cloud"; + Config.cloud_unique_id = ""; + + TOlapTableSink thriftSink = new TOlapTableSink(); + thriftSink.setDbId(100L); + OlapTableSink sink = new OlapTableSink(table, null, Collections.singletonList(10L)); + TOlapTableLocationParam location = sink.initLocationParam(thriftSink); + + Assertions.assertEquals(Collections.singletonList(2L), + location.getTablets().get(0).getNodeIds()); + Mockito.verify(systemInfoService, Mockito.never()).checkBackendLoadAvailable(Mockito.anyLong()); + } finally { + Config.deploy_mode = originalDeployMode; + Config.cloud_unique_id = originalCloudUniqueId; + } + } + + private OlapTable mockTableWithSingleTablet(Tablet tablet, int loadRequiredReplicaNum) { + PartitionInfo partitionInfo = Mockito.mock(PartitionInfo.class); + Partition partition = Mockito.mock(Partition.class); + MaterializedIndex index = Mockito.mock(MaterializedIndex.class); + OlapTable table = Mockito.mock(OlapTable.class); + Mockito.when(partitionInfo.enableAutomaticPartition()).thenReturn(false); + Mockito.when(table.getPartitionInfo()).thenReturn(partitionInfo); + Mockito.when(table.getPartition(10L)).thenReturn(partition); + Mockito.when(table.getLoadRequiredReplicaNum(10L)).thenReturn(loadRequiredReplicaNum); + Mockito.when(partition.getMaterializedIndices(IndexExtState.ALL, true)) + .thenReturn(Collections.singletonList(index)); + Mockito.when(index.getTablets()).thenReturn(Collections.singletonList(tablet)); + return table; + } + @Test public void testCreateDummyLocationUsesLoadAvailableBackendInCurrentComputeGroup() throws Exception { SystemInfoService systemInfoService = Mockito.mock(SystemInfoService.class); diff --git a/fe/fe-core/src/test/java/org/apache/doris/qe/LoadContextTest.java b/fe/fe-core/src/test/java/org/apache/doris/qe/LoadContextTest.java new file mode 100644 index 00000000000000..6515390952ceda --- /dev/null +++ b/fe/fe-core/src/test/java/org/apache/doris/qe/LoadContextTest.java @@ -0,0 +1,81 @@ +// 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. + +package org.apache.doris.qe; + +import org.apache.doris.common.Config; +import org.apache.doris.thrift.TErrorTabletInfo; +import org.apache.doris.thrift.TTabletCommitInfo; +import org.apache.doris.transaction.TabletCommitInfo; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import java.util.Arrays; +import java.util.List; + +public class LoadContextTest { + @Test + public void testErrorTabletsAddSharedNothingParticipationMarkers() { + String oldDeployMode = Config.deploy_mode; + String oldCloudUniqueId = Config.cloud_unique_id; + int oldMaxErrorTablets = Config.max_error_tablet_of_broker_load; + try { + Config.deploy_mode = ""; + Config.cloud_unique_id = ""; + Config.max_error_tablet_of_broker_load = 0; + LoadContext context = new LoadContext(); + TErrorTabletInfo first = new TErrorTabletInfo(); + first.setTabletId(100L); + first.setMsg("first destination failed"); + TErrorTabletInfo second = new TErrorTabletInfo(); + second.setTabletId(100L); + second.setMsg("second destination failed"); + + context.updateErrorTabletInfos(Arrays.asList(first, second)); + List commitInfos = context.getCommitInfos(); + Assertions.assertEquals(1, commitInfos.size()); + Assertions.assertEquals(100L, commitInfos.get(0).getTabletId()); + Assertions.assertEquals(TabletCommitInfo.PARTITION_INVOLVEMENT_BACKEND_ID, + commitInfos.get(0).getBackendId()); + Assertions.assertTrue(context.getErrorTabletInfos().isEmpty()); + } finally { + Config.deploy_mode = oldDeployMode; + Config.cloud_unique_id = oldCloudUniqueId; + Config.max_error_tablet_of_broker_load = oldMaxErrorTablets; + } + } + + @Test + public void testCloudErrorTabletsDoNotAddParticipationMarkers() { + String oldDeployMode = Config.deploy_mode; + String oldCloudUniqueId = Config.cloud_unique_id; + try { + Config.deploy_mode = "cloud"; + Config.cloud_unique_id = "cloud-test"; + LoadContext context = new LoadContext(); + TErrorTabletInfo error = new TErrorTabletInfo(); + error.setTabletId(100L); + + context.updateErrorTabletInfos(Arrays.asList(error)); + Assertions.assertTrue(context.getCommitInfos().isEmpty()); + } finally { + Config.deploy_mode = oldDeployMode; + Config.cloud_unique_id = oldCloudUniqueId; + } + } +} diff --git a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java index efa76702539f0b..601f8b976c8b72 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/service/FrontendServiceImplTest.java @@ -31,12 +31,14 @@ import org.apache.doris.common.FeConstants; import org.apache.doris.common.util.DatasourcePrintableMap; import org.apache.doris.datasource.InternalCatalog; +import org.apache.doris.insertoverwrite.InsertOverwriteManager; import org.apache.doris.mysql.authenticate.TestLogAppender; import org.apache.doris.nereids.parser.NereidsParser; import org.apache.doris.nereids.trees.plans.commands.Command; import org.apache.doris.nereids.trees.plans.commands.CreateDatabaseCommand; import org.apache.doris.nereids.trees.plans.logical.LogicalPlan; import org.apache.doris.qe.StmtExecutor; +import org.apache.doris.system.Backend; import org.apache.doris.tablefunction.BackendsTableValuedFunction; import org.apache.doris.thrift.TBackendsMetadataParams; import org.apache.doris.thrift.TCheckAuthRequest; @@ -61,6 +63,8 @@ import org.apache.doris.thrift.TPrivilegeCtrl; import org.apache.doris.thrift.TPrivilegeHier; import org.apache.doris.thrift.TPrivilegeType; +import org.apache.doris.thrift.TReplacePartitionRequest; +import org.apache.doris.thrift.TReplacePartitionResult; import org.apache.doris.thrift.TRollbackTxnRequest; import org.apache.doris.thrift.TSchemaTableName; import org.apache.doris.thrift.TSchemaTableRequestParams; @@ -92,6 +96,7 @@ import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; @@ -660,6 +665,88 @@ private static TCreatePartitionRequest createPartitionRequest(Database db, OlapT return request; } + @Test + public void testCreatePartitionRejectsEmptyLoadLocation() throws Exception { + createTable("CREATE TABLE test.partition_no_load_backend (city VARCHAR(20) NOT NULL, id INT) " + + "DUPLICATE KEY(city, id) AUTO PARTITION BY LIST (city) () " + + "DISTRIBUTED BY HASH(id) BUCKETS 1 PROPERTIES('replication_num' = '1')"); + Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException("test"); + OlapTable table = (OlapTable) db.getTableOrAnalysisException("partition_no_load_backend"); + FrontendServiceImpl service = new FrontendServiceImpl(exeEnv); + TCreatePartitionResult successfulResult = service + .createPartition(createPartitionRequest(db, table, "SHANGHAI")); + Assertions.assertEquals(TStatusCode.OK, successfulResult.getStatus().getStatusCode()); + Assertions.assertEquals(1, successfulResult.getPartitions().get(0).getTotalReplicaNum()); + Assertions.assertEquals(1, successfulResult.getPartitions().get(0).getLoadRequiredReplicaNum()); + + Map originalLoadDisabled = disableLoadOnAllBackends(); + try { + TCreatePartitionResult result = service.createPartition(createPartitionRequest(db, table, "BEIJING")); + + Assertions.assertEquals(TStatusCode.RUNTIME_ERROR, result.getStatus().getStatusCode()); + Assertions.assertTrue(result.getStatus().getErrorMsgs().get(0) + .contains("no load available backend for tablet")); + } finally { + restoreLoadDisabled(originalLoadDisabled); + } + } + + @Test + public void testReplacePartitionRejectsEmptyLoadLocation() throws Exception { + createTable("CREATE TABLE test.replace_partition_no_load_backend (id INT, city INT) " + + "DUPLICATE KEY(id) PARTITION BY RANGE(city) " + + "(PARTITION p1 VALUES LESS THAN (10), PARTITION p2 VALUES LESS THAN (20)) " + + "DISTRIBUTED BY HASH(id) BUCKETS 1 " + + "PROPERTIES('replication_num' = '1')"); + Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException("test"); + OlapTable table = (OlapTable) db.getTableOrAnalysisException("replace_partition_no_load_backend"); + InsertOverwriteManager overwriteManager = Env.getCurrentEnv().getInsertOverwriteManager(); + long groupId = overwriteManager.registerTaskGroup(table); + FrontendServiceImpl service = new FrontendServiceImpl(exeEnv); + + TReplacePartitionRequest successfulRequest = new TReplacePartitionRequest(); + successfulRequest.setDbId(db.getId()); + successfulRequest.setTableId(table.getId()); + successfulRequest.setPartitionIds(Collections.singletonList(table.getPartition("p1").getId())); + successfulRequest.setOverwriteGroupId(groupId); + TReplacePartitionResult successfulResult = service.replacePartition(successfulRequest); + Assertions.assertEquals(TStatusCode.OK, successfulResult.getStatus().getStatusCode()); + Assertions.assertEquals(1, successfulResult.getPartitions().get(0).getTotalReplicaNum()); + Assertions.assertEquals(1, successfulResult.getPartitions().get(0).getLoadRequiredReplicaNum()); + + Map originalLoadDisabled = disableLoadOnAllBackends(); + try { + TReplacePartitionRequest request = new TReplacePartitionRequest(); + request.setDbId(db.getId()); + request.setTableId(table.getId()); + request.setPartitionIds(Collections.singletonList(table.getPartition("p2").getId())); + request.setOverwriteGroupId(groupId); + TReplacePartitionResult result = service.replacePartition(request); + + Assertions.assertEquals(TStatusCode.RUNTIME_ERROR, result.getStatus().getStatusCode()); + Assertions.assertTrue(result.getStatus().getErrorMsgs().get(0) + .contains("no load available backend for tablet")); + } finally { + restoreLoadDisabled(originalLoadDisabled); + overwriteManager.taskGroupFail(groupId); + } + } + + private Map disableLoadOnAllBackends() { + Map originalLoadDisabled = new HashMap<>(); + for (long backendId : Env.getCurrentSystemInfo().getAllBackendIds(false)) { + Backend backend = Env.getCurrentSystemInfo().getBackend(backendId); + originalLoadDisabled.put(backendId, backend.isLoadDisabled()); + backend.setLoadDisabled(true); + } + return originalLoadDisabled; + } + + private void restoreLoadDisabled(Map originalLoadDisabled) { + originalLoadDisabled.forEach((backendId, disabled) -> + Env.getCurrentSystemInfo().getBackend(backendId).setLoadDisabled(disabled)); + } + @Test public void testCreatePartitionWithChangelog() throws Exception { String createOlapTblStmt = "CREATE TABLE test.partition_range_with_changelog(\n" diff --git a/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java b/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java index 3d41243955cae0..74627a06d6c4bd 100644 --- a/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java +++ b/fe/fe-core/src/test/java/org/apache/doris/transaction/DatabaseTransactionMgrTest.java @@ -234,6 +234,28 @@ public Map addTransactionToTransactionMgr() throws UserException { return labelToTxnId; } + @Test + public void testInvolvedTabletMarkerRequiresSuccessfulReplica() throws UserException { + FakeEnv.setEnv(masterEnv); + Database db = masterEnv.getInternalCatalog().getDbOrMetaException(CatalogTestUtil.testDbId1); + OlapTable failedTable = (OlapTable) db.getTableOrMetaException(CatalogTestUtil.testTableId1); + OlapTable healthyTable = (OlapTable) db.getTableOrMetaException(CatalogTestUtil.testTableId2); + long transactionId = masterTransMgr.beginTransaction(CatalogTestUtil.testDbId1, + Lists.newArrayList(CatalogTestUtil.testTableId1, CatalogTestUtil.testTableId2), + "involved_tablet_marker", transactionSource, LoadJobSourceType.FRONTEND, + Config.stream_load_default_timeout_second); + List commitInfos = Lists.newArrayList( + new TabletCommitInfo(CatalogTestUtil.testTabletId1, + TabletCommitInfo.PARTITION_INVOLVEMENT_BACKEND_ID), + new TabletCommitInfo(CatalogTestUtil.testTabletId2, CatalogTestUtil.testBackendId1)); + + TabletQuorumFailedException exception = Assertions.assertThrows(TabletQuorumFailedException.class, + () -> masterTransMgr.commitTransactionWithoutLock(CatalogTestUtil.testDbId1, + Lists.newArrayList(failedTable, healthyTable), transactionId, commitInfos, null)); + Assertions.assertTrue(exception.getMessage().contains("tablet " + CatalogTestUtil.testTabletId1 + + " succ replica num 0")); + } + @Test public void testNormal() throws UserException { DatabaseTransactionMgr masterDbTransMgr = masterTransMgr.getDatabaseTransactionMgr(CatalogTestUtil.testDbId1); diff --git a/regression-test/suites/fault_injection_p0/test_disable_load_selected_replica_failure.groovy b/regression-test/suites/fault_injection_p0/test_disable_load_selected_replica_failure.groovy new file mode 100644 index 00000000000000..822359e130c3bd --- /dev/null +++ b/regression-test/suites/fault_injection_p0/test_disable_load_selected_replica_failure.groovy @@ -0,0 +1,144 @@ +// 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.suite.ClusterOptions +import org.apache.doris.regression.util.NodeType + +suite("test_disable_load_selected_replica_failure", "docker") { + def options = new ClusterOptions() + options.cloudMode = false + options.feNum = 1 + options.beNum = 5 + options.feConfigs += ["disable_tablet_scheduler=true"] + options.enableDebugPoints() + + docker(options) { + def backendIPs = [:] + def backendPorts = [:] + getBackendIpHttpPort(backendIPs, backendPorts) + def backendIds = backendIPs.keySet().sort { it.toLong() } + def sharedBackend = backendIds[0] + def disabledBackends = backendIds[1..2] + def tags = ["shared", "left", "left", "right", "right"] + backendIds.eachWithIndex { backendId, i -> + sql """ALTER SYSTEM MODIFY BACKEND "${backendId}" SET ("tag.location" = "${tags[i]}")""" + } + + def originalMemtableOnSink = sql("select @@enable_memtable_on_sink_node")[0][0] + def legacyPoint = "LoadChannel.add_batch.failed" + def v2Point = "LoadStreamWriter.close.cancelled" + def debugPoint = GetDebugPoint() + try { + sql "set enable_memtable_on_sink_node = false" + sql "drop table if exists disable_load_empty_auto_partition" + sql """ + create table disable_load_empty_auto_partition (city varchar(20), id int) + duplicate key(city, id) + auto partition by list(city) () + distributed by hash(id) buckets 1 + properties ( + "replication_allocation" = + "tag.location.shared: 1, tag.location.left: 1, tag.location.right: 1" + ) + """ + // The synthetic partition/tablet 0 has one destination even though real partitions require two. + sql "insert into disable_load_empty_auto_partition values ('BEIJING', 1)" + sql """select assert_true(count(*) = 1 and min(id) = 1) + from disable_load_empty_auto_partition""" + + sql "drop table if exists disable_load_selected_failure_legacy" + sql """ + create table disable_load_selected_failure_legacy (k int, v int) + duplicate key(k) + partition by range(k) ( + partition p_a values less than ("10"), + partition p_b values less than ("30") + ("replication_allocation" = "tag.location.shared: 1, tag.location.right: 2") + ) + distributed by hash(k) buckets 1 + properties ( + "replication_allocation" = "tag.location.shared: 1, tag.location.left: 2", + "min_load_replica_num" = "1" + ) + """ + sql "drop table if exists disable_load_selected_failure_v2" + sql """ + create table disable_load_selected_failure_v2 (k int, v int) + duplicate key(k) + partition by range(k) ( + partition p_a values less than ("10"), + partition p_b values less than ("30") + ("replication_allocation" = "tag.location.shared: 1, tag.location.right: 2") + ) + distributed by hash(k) buckets 1 + properties ( + "replication_allocation" = "tag.location.shared: 1, tag.location.left: 2", + "min_load_replica_num" = "1" + ) + """ + + disabledBackends.each { backendId -> + sql """ALTER SYSTEM MODIFY BACKEND "${backendId}" SET ("disable_load" = "true")""" + } + + // Seed B before enabling the shared-backend fault. + sql "insert into disable_load_selected_failure_legacy values (21, 100)" + debugPoint.enableDebugPoint(backendIPs[sharedBackend], + backendPorts[sharedBackend] as int, NodeType.BE, legacyPoint) + test { + sql "insert into disable_load_selected_failure_legacy values (1, 200), (22, 200)" + exception "fault injection" + } + sql """select assert_true(count(*) = 1 and min(k) = 21 and min(v) = 100) + from disable_load_selected_failure_legacy""" + debugPoint.disableDebugPoint(backendIPs[sharedBackend], + backendPorts[sharedBackend] as int, NodeType.BE, legacyPoint) + sql "insert into disable_load_selected_failure_legacy values (1, 200), (22, 200)" + sql """select assert_true(count(*) = 3 and sum(k) = 44 and sum(v) = 500) + from disable_load_selected_failure_legacy""" + + sql "set enable_memtable_on_sink_node = true" + // Seed B before enabling the V2 late-close fault. + sql "insert into disable_load_selected_failure_v2 values (21, 100)" + debugPoint.enableDebugPoint(backendIPs[sharedBackend], + backendPorts[sharedBackend] as int, NodeType.BE, v2Point) + test { + sql "insert into disable_load_selected_failure_v2 values (1, 200), (22, 200)" + exception "flush segment failed" + } + sql """select assert_true(count(*) = 1 and min(k) = 21 and min(v) = 100) + from disable_load_selected_failure_v2""" + debugPoint.disableDebugPoint(backendIPs[sharedBackend], + backendPorts[sharedBackend] as int, NodeType.BE, v2Point) + sql "insert into disable_load_selected_failure_v2 values (1, 200), (22, 200)" + sql """select assert_true(count(*) = 3 and sum(k) = 44 and sum(v) = 500) + from disable_load_selected_failure_v2""" + } finally { + try { + debugPoint.disableDebugPoint(backendIPs[sharedBackend], + backendPorts[sharedBackend] as int, NodeType.BE, legacyPoint) + debugPoint.disableDebugPoint(backendIPs[sharedBackend], + backendPorts[sharedBackend] as int, NodeType.BE, v2Point) + } finally { + disabledBackends.each { backendId -> + sql """ALTER SYSTEM MODIFY BACKEND "${backendId}" SET ("disable_load" = "false")""" + } + sql "set enable_memtable_on_sink_node = '${originalMemtableOnSink}'" + } + } + } +}