Skip to content

[fix](fe) Exclude load-disabled backends from load plans #69554

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
HonestManXin wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[fix](fe) Exclude load-disabled backends from load plans #69554

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions be/src/exec/sink/load_stream_map_pool.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

#include "exec/sink/load_stream_map_pool.h"

#include <algorithm>

#include "util/debug_points.h"

namespace doris {
Expand Down Expand Up @@ -99,6 +101,64 @@ void LoadStreamMap::save_tablets_to_commit(int64_t dst_id,
}
}

void LoadStreamMap::append_partition_involvement_commit_infos(
std::vector<TTabletCommitInfo>& tablet_commit_infos) {
constexpr int64_t PARTITION_INVOLVEMENT_BACKEND_ID = -1;
constexpr int64_t DUMMY_TABLET_ID = 0;
std::unordered_map<int64_t, int64_t> partition_to_tablet;
{
std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> 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) {
Expand Down
15 changes: 15 additions & 0 deletions be/src/exec/sink/load_stream_map_pool.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<int64_t> version_gap_backends;
};

LoadStreamMap(UniqueId load_id, int64_t src_id, int num_streams, int num_use,
LoadStreamMapPool* pool);

Expand All @@ -85,6 +92,13 @@ class LoadStreamMap {

void save_tablets_to_commit(int64_t dst_id, const std::vector<PTabletID>& tablets_to_commit);

void append_partition_involvement_commit_infos(
std::vector<TTabletCommitInfo>& 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<int64_t, int32_t>& segments_for_tablet) {
_segments_for_tablet.insert(segments_for_tablet.cbegin(), segments_for_tablet.cend());
}
Expand Down Expand Up @@ -117,6 +131,7 @@ class LoadStreamMap {
std::atomic<int> _num_incremental_streams;
std::mutex _mutex;
std::unordered_map<int64_t, std::shared_ptr<LoadStreamStubs>> _streams_for_node;
std::unordered_map<int64_t, TabletReplicaInfo> _tablet_replica_infos;
LoadStreamMapPool* _pool = nullptr;
std::shared_ptr<IndexToTabletSchema> _tablet_schema_for_index;
std::shared_ptr<IndexToEnableMoW> _enable_unique_mow_for_index;
Expand Down
190 changes: 150 additions & 40 deletions be/src/exec/sink/writer/vtablet_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,6 +46,7 @@
#include <vector>

#include "cloud/config.h"
#include "common/check.h"
#include "common/compiler_util.h" // IWYU pragma: keep
#include "common/config.h"
#include "common/logging.h"
Expand Down Expand Up @@ -92,6 +93,8 @@
namespace doris {
class TExpr;

static constexpr int64_t V1_DUMMY_TABLET_ID = 0;

bvar::Adder<int64_t> g_sink_write_bytes;
bvar::PerSecond<bvar::Adder<int64_t>> g_sink_write_bytes_per_second("sink_throughput_byte",
&g_sink_write_bytes, 60);
Expand Down Expand Up @@ -136,6 +139,17 @@ Status IndexChannel::init(RuntimeState* state, const std::vector<TTabletWithPart
if (tablet_locations == nullptr) {
return Status::InternalError("unknown tablet, tablet_id={}", tablet.tablet_id);
}
int selected_replicas_num = cast_set<int>(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) {
Comment thread
HonestManXin marked this conversation as resolved.
// 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<std::shared_ptr<VNodeChannel>> channels;
// For tablet, deal with its' all replica (in some node).
for (auto& replica_node_id : tablet_locations->node_ids) {
Expand Down Expand Up @@ -174,7 +188,11 @@ Status IndexChannel::init(RuntimeState* state, const std::vector<TTabletWithPart
}
}
channels.push_back(channel);
_tablets_by_channel[replica_node_id].insert(tablet.tablet_id);
if (tablet.tablet_id != V1_DUMMY_TABLET_ID) [[likely]] {
// Tablet 0 only bootstraps an empty automatic table. It receives no rows and must
// not consume the real tablets' failure allowance before partitions are created.
_tablets_by_channel[replica_node_id].insert(tablet.tablet_id);
Comment thread
HonestManXin marked this conversation as resolved.
}
}
_channels_by_tablet.emplace(tablet.tablet_id, std::move(channels));
}
Expand All @@ -195,50 +213,75 @@ void IndexChannel::mark_as_failed(const VNodeChannel* node_channel, const std::s
LOG(INFO) << "mark node_id:" << node_channel->channel_info() << " tablet_id: " << tablet_id
<< " as failed, err: " << err;
auto node_id = node_channel->node_id();
if (tablet_id != -1) {
std::lock_guard<std::mutex> 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<ErrorCode::INTERNAL_ERROR, false>(
_failed_channels_msgs[tablet_id]);
}
return;
}

const auto& it = _tablets_by_channel.find(node_id);
if (it == _tablets_by_channel.end()) {
std::lock_guard<std::mutex> 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<std::mutex> 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<ErrorCode::INTERNAL_ERROR, false>(
_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<ErrorCode::INTERNAL_ERROR, false>(
_failed_channels_msgs[tablet_id]);
}
std::lock_guard<std::mutex> 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<ErrorCode::INTERNAL_ERROR, false>(
_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));
Comment thread
HonestManXin marked this conversation as resolved.
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() {
Expand Down Expand Up @@ -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()) {
Expand Down Expand Up @@ -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!";
}
Expand Down Expand Up @@ -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<int>(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<TTabletCommitInfo> 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) {
Expand Down Expand Up @@ -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<int64_t, AddBatchCounter> node_add_batch_counter_map;
Expand Down
Loading
Loading