github-actions[bot] commented on code in PR #66623:
URL: https://github.com/apache/doris/pull/66623#discussion_r3756114765
##########
be/src/service/internal_service.cpp:
##########
@@ -1973,309 +1966,20 @@ void
PInternalService::hand_shake(google::protobuf::RpcController* controller,
response->mutable_status()->set_status_code(0);
}
-constexpr char HttpProtocol[] = "http://";;
-constexpr char DownloadApiPath[] = "/api/_tablet/_download?token=";
-constexpr char FileParam[] = "&file=";
-
-static std::string construct_url(const std::string& host_port, const
std::string& token,
- const std::string& path) {
- return fmt::format("{}{}{}{}{}{}", HttpProtocol, host_port,
DownloadApiPath, token, FileParam,
- path);
-}
-
-static Status download_file_action(std::string& remote_file_url, std::string&
local_file_path,
- uint64_t estimate_timeout, uint64_t
file_size) {
- auto download_cb = [remote_file_url, estimate_timeout, local_file_path,
- file_size](HttpClient* client) {
- RETURN_IF_ERROR(client->init(remote_file_url));
- client->set_timeout_ms(estimate_timeout * 1000);
- RETURN_IF_ERROR(client->download(local_file_path));
-
- if (file_size > 0) {
- // Check file length
- uint64_t local_file_size =
std::filesystem::file_size(local_file_path);
- if (local_file_size != file_size) {
- LOG(WARNING) << "failed to pull rowset for slave replica.
download file "
- "length error"
- << ", remote_path=" << remote_file_url << ",
file_size=" << file_size
- << ", local_file_size=" << local_file_size;
- return Status::InternalError("downloaded file size is not
equal");
- }
- }
-
- return io::global_local_filesystem()->permission(local_file_path,
-
io::LocalFileSystem::PERMS_OWNER_RW);
- };
- return HttpClient::execute_with_retry(DOWNLOAD_FILE_MAX_RETRY, 1,
download_cb);
-}
-
void PInternalServiceImpl::request_slave_tablet_pull_rowset(
google::protobuf::RpcController* controller, const
PTabletWriteSlaveRequest* request,
PTabletWriteSlaveResult* response, google::protobuf::Closure* done) {
brpc::ClosureGuard closure_guard(done);
- const RowsetMetaPB& rowset_meta_pb = request->rowset_meta();
- const std::string& rowset_path = request->rowset_path();
- google::protobuf::Map<int64_t, int64_t> segments_size =
request->segments_size();
- google::protobuf::Map<int64_t, PTabletWriteSlaveRequest_IndexSizeMap>
indices_size =
- request->inverted_indices_size();
- std::string host = request->host();
- int64_t http_port = request->http_port();
- int64_t brpc_port = request->brpc_port();
- std::string token = request->token();
- int64_t node_id = request->node_id();
- bool ret = _heavy_work_pool.try_offer([rowset_meta_pb, host, brpc_port,
node_id, segments_size,
- indices_size, http_port, token,
rowset_path, this]() {
- TabletSharedPtr tablet = _engine.tablet_manager()->get_tablet(
- rowset_meta_pb.tablet_id(),
rowset_meta_pb.tablet_schema_hash());
- if (tablet == nullptr) {
- LOG(WARNING) << "failed to pull rowset for slave replica. tablet ["
- << rowset_meta_pb.tablet_id()
- << "] is not exist. txn_id=" <<
rowset_meta_pb.txn_id();
- _response_pull_slave_rowset(host, brpc_port,
rowset_meta_pb.txn_id(),
- rowset_meta_pb.tablet_id(), node_id,
false);
- return;
- }
-
- RowsetMetaSharedPtr rowset_meta(new RowsetMeta());
- std::string rowset_meta_str;
- bool ret = rowset_meta_pb.SerializeToString(&rowset_meta_str);
- if (!ret) {
- LOG(WARNING) << "failed to pull rowset for slave replica.
serialize rowset meta "
- "failed. rowset_id="
- << rowset_meta_pb.rowset_id()
- << ", tablet_id=" << rowset_meta_pb.tablet_id()
- << ", txn_id=" << rowset_meta_pb.txn_id();
- _response_pull_slave_rowset(host, brpc_port,
rowset_meta_pb.txn_id(),
- rowset_meta_pb.tablet_id(), node_id,
false);
- return;
- }
- bool parsed = rowset_meta->init(rowset_meta_str);
- if (!parsed) {
- LOG(WARNING) << "failed to pull rowset for slave replica. parse
rowset meta string "
- "failed. rowset_id="
- << rowset_meta_pb.rowset_id()
- << ", tablet_id=" << rowset_meta_pb.tablet_id()
- << ", txn_id=" << rowset_meta_pb.txn_id();
- // return false will break meta iterator, return true to skip this
error
- _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
- rowset_meta->tablet_id(), node_id,
false);
- return;
- }
- RowsetId remote_rowset_id = rowset_meta->rowset_id();
- // change rowset id because it maybe same as other local rowset
- RowsetId new_rowset_id = _engine.next_rowset_id();
- auto pending_rs_guard =
_engine.pending_local_rowsets().add(new_rowset_id);
- rowset_meta->set_rowset_id(new_rowset_id);
- rowset_meta->set_tablet_uid(tablet->tablet_uid());
- VLOG_CRITICAL << "succeed to init rowset meta for slave replica.
rowset_id="
- << rowset_meta->rowset_id() << ", tablet_id=" <<
rowset_meta->tablet_id()
- << ", txn_id=" << rowset_meta->txn_id();
-
- auto tablet_scheme = rowset_meta->tablet_schema();
- for (const auto& segment : segments_size) {
- uint64_t file_size = segment.second;
- uint64_t estimate_timeout = file_size /
config::download_low_speed_limit_kbps / 1024;
- if (estimate_timeout < config::download_low_speed_time) {
- estimate_timeout = config::download_low_speed_time;
- }
-
- std::string remote_file_path =
- local_segment_path(rowset_path,
remote_rowset_id.to_string(), segment.first);
- std::string remote_file_url =
- construct_url(get_host_port(host, http_port), token,
remote_file_path);
-
- std::string local_file_path = local_segment_path(
- tablet->tablet_path(),
rowset_meta->rowset_id().to_string(), segment.first);
-
- auto st = download_file_action(remote_file_url, local_file_path,
estimate_timeout,
- file_size);
- if (!st.ok()) {
- LOG(WARNING) << "failed to pull rowset for slave replica.
failed to download "
- "file. url="
- << remote_file_url << ", local_path=" <<
local_file_path
- << ", txn_id=" << rowset_meta->txn_id();
- _response_pull_slave_rowset(host, brpc_port,
rowset_meta->txn_id(),
- rowset_meta->tablet_id(), node_id,
false);
- return;
- }
- VLOG_CRITICAL << "succeed to download file for slave replica.
url=" << remote_file_url
- << ", local_path=" << local_file_path
- << ", txn_id=" << rowset_meta->txn_id();
- if (indices_size.find(segment.first) != indices_size.end()) {
- PTabletWriteSlaveRequest_IndexSizeMap segment_indices_size =
- indices_size.at(segment.first);
-
- for (auto index_size : segment_indices_size.index_sizes()) {
- auto index_id = index_size.indexid();
- auto size = index_size.size();
- auto suffix_path = index_size.suffix_path();
- std::string remote_inverted_index_file;
- std::string local_inverted_index_file;
- std::string remote_inverted_index_file_url;
- if (tablet_scheme->get_inverted_index_storage_format() ==
- InvertedIndexStorageFormatPB::V1) {
- remote_inverted_index_file =
-
InvertedIndexDescriptor::get_index_file_path_v1(
-
InvertedIndexDescriptor::get_index_file_path_prefix(
- remote_file_path),
- index_id, suffix_path);
- remote_inverted_index_file_url = construct_url(
- get_host_port(host, http_port), token,
remote_inverted_index_file);
-
- local_inverted_index_file =
InvertedIndexDescriptor::get_index_file_path_v1(
-
InvertedIndexDescriptor::get_index_file_path_prefix(
- local_file_path),
- index_id, suffix_path);
- } else {
- remote_inverted_index_file =
-
InvertedIndexDescriptor::get_index_file_path_v2(
-
InvertedIndexDescriptor::get_index_file_path_prefix(
- remote_file_path));
- remote_inverted_index_file_url = construct_url(
- get_host_port(host, http_port), token,
remote_inverted_index_file);
-
- local_inverted_index_file =
InvertedIndexDescriptor::get_index_file_path_v2(
-
InvertedIndexDescriptor::get_index_file_path_prefix(
- local_file_path));
- }
- st = download_file_action(remote_inverted_index_file_url,
- local_inverted_index_file,
estimate_timeout, size);
- if (!st.ok()) {
- LOG(WARNING) << "failed to pull rowset for slave
replica. failed to "
- "download "
- "file. url="
- << remote_inverted_index_file_url
- << ", local_path=" <<
local_inverted_index_file
- << ", txn_id=" << rowset_meta->txn_id();
- _response_pull_slave_rowset(host, brpc_port,
rowset_meta->txn_id(),
- rowset_meta->tablet_id(),
node_id, false);
- return;
- }
-
- VLOG_CRITICAL
- << "succeed to download inverted index file for
slave replica. url="
- << remote_inverted_index_file_url
- << ", local_path=" << local_inverted_index_file
- << ", txn_id=" << rowset_meta->txn_id();
- }
- }
- }
-
- RowsetSharedPtr rowset;
- Status create_status = RowsetFactory::create_rowset(
- tablet->tablet_schema(), tablet->tablet_path(), rowset_meta,
&rowset);
- if (!create_status) {
- LOG(WARNING) << "failed to create rowset from rowset meta for
slave replica"
- << ". rowset_id: " << rowset_meta->rowset_id()
- << ", rowset_type: " << rowset_meta->rowset_type()
- << ", rowset_state: " << rowset_meta->rowset_state()
- << ", tablet_id=" << rowset_meta->tablet_id()
- << ", txn_id=" << rowset_meta->txn_id();
- _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
- rowset_meta->tablet_id(), node_id,
false);
- return;
- }
- if (rowset_meta->rowset_state() != RowsetStatePB::COMMITTED) {
- LOG(WARNING) << "could not commit txn for slave replica because
master rowset state is "
- "not committed, rowset_state="
- << rowset_meta->rowset_state()
- << ", tablet_id=" << rowset_meta->tablet_id()
- << ", txn_id=" << rowset_meta->txn_id();
- _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
- rowset_meta->tablet_id(), node_id,
false);
- return;
- }
- Status commit_txn_status = _engine.txn_manager()->commit_txn(
- tablet->data_dir()->get_meta(), rowset_meta->partition_id(),
rowset_meta->txn_id(),
- rowset_meta->tablet_id(), tablet->tablet_uid(),
rowset_meta->load_id(), rowset,
- std::move(pending_rs_guard), false);
- if (!commit_txn_status &&
!commit_txn_status.is<PUSH_TRANSACTION_ALREADY_EXIST>()) {
- LOG(WARNING) << "failed to add committed rowset for slave replica.
rowset_id="
- << rowset_meta->rowset_id() << ", tablet_id=" <<
rowset_meta->tablet_id()
- << ", txn_id=" << rowset_meta->txn_id();
- _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
- rowset_meta->tablet_id(), node_id,
false);
- return;
- }
- VLOG_CRITICAL << "succeed to pull rowset for slave replica.
successfully to add committed "
- "rowset: "
- << rowset_meta->rowset_id()
- << " to tablet, tablet_id=" << rowset_meta->tablet_id()
- << ", schema_hash=" << rowset_meta->tablet_schema_hash()
- << ", txn_id=" << rowset_meta->txn_id();
- _response_pull_slave_rowset(host, brpc_port, rowset_meta->txn_id(),
- rowset_meta->tablet_id(), node_id, true);
- });
- if (!ret) {
- offer_failed(response, closure_guard.release(), _heavy_work_pool);
- return;
- }
- Status::OK().to_protobuf(response->mutable_status());
-}
-
-void PInternalServiceImpl::_response_pull_slave_rowset(const std::string&
remote_host,
- int64_t brpc_port,
int64_t txn_id,
- int64_t tablet_id,
int64_t node_id,
- bool is_succeed) {
- std::shared_ptr<PBackendService_Stub> stub =
-
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(remote_host,
-
brpc_port);
- if (stub == nullptr) {
- LOG(WARNING) << "failed to response result of slave replica to master
replica. get rpc "
- "stub failed, master host="
- << remote_host << ", port=" << brpc_port << ",
tablet_id=" << tablet_id
- << ", txn_id=" << txn_id;
- return;
- }
-
- auto request = std::make_shared<PTabletWriteSlaveDoneRequest>();
- request->set_txn_id(txn_id);
- request->set_tablet_id(tablet_id);
- request->set_node_id(node_id);
- request->set_is_succeed(is_succeed);
- auto pull_rowset_callback =
DummyBrpcCallback<PTabletWriteSlaveDoneResult>::create_shared();
- auto closure = AutoReleaseClosure<
- PTabletWriteSlaveDoneRequest,
-
DummyBrpcCallback<PTabletWriteSlaveDoneResult>>::create_unique(request,
-
pull_rowset_callback);
-
closure->cntl_->set_timeout_ms(config::slave_replica_writer_rpc_timeout_sec *
1000);
- closure->cntl_->ignore_eovercrowded();
- stub->response_slave_tablet_pull_rowset(closure->cntl_.get(),
closure->request_.get(),
- closure->response_.get(),
closure.get());
- closure.release();
-
- pull_rowset_callback->join();
- if (pull_rowset_callback->cntl_->Failed()) {
- LOG(WARNING) << "failed to response result of slave replica to master
replica, error="
- << berror(pull_rowset_callback->cntl_->ErrorCode())
- << ", error_text=" <<
pull_rowset_callback->cntl_->ErrorText()
- << ", master host: " << remote_host << ", tablet_id=" <<
tablet_id
- << ", txn_id=" << txn_id;
- }
- VLOG_CRITICAL << "succeed to response the result of slave replica pull
rowset to master "
- "replica. master host: "
- << remote_host << ". is_succeed=" << is_succeed << ",
tablet_id=" << tablet_id
- << ", slave server=" << node_id << ", txn_id=" << txn_id;
+ Status::NotSupported("single replica load has been removed")
Review Comment:
During a BE-first rolling upgrade, an old FE can choose an old BE as the
single-replica master and an upgraded BE as a slave. The old `DeltaWriter`
records that slave in `_unfinished_slave_node`, but after this RPC it checks
only `cntl_->Failed()`; it does not inspect `PTabletWriteSlaveResult.status`,
and it clears the entry only on transport failure or the later
`response_slave_tablet_pull_rowset` callback. This handler completes BRPC
successfully with `NotSupported` and sends no callback, so
`TabletsChannel::close` polls until roughly 90% of the load timeout and may
then fail quorum. Please make the retained RPC complete the legacy callback
contract (or fail in a way the old caller recognizes) and cover the
old-master/new-slave placement with a mixed-version test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java:
##########
@@ -2138,10 +2137,6 @@ public boolean isEnableHboNonStrictMatchingMode() {
@VarAttrDef.VarAttr(name = SESSION_CONTEXT, needForward = true)
public String sessionContext = "";
- @VarAttrDef.VarAttr(name = ENABLE_SINGLE_REPLICA_INSERT,
- needForward = true, varType = VariableAnnotation.EXPERIMENTAL)
- public boolean enableSingleReplicaInsert = false;
Review Comment:
Doris keeps `VariableMgr.REMOVED_SESSION_VAR_NAMES` specifically so removed
variables still no-op/read safely in user scripts, JDBC initialization, and
rolling upgrades. After deleting this field, neither
`enable_single_replica_insert` nor its previously accepted
`experimental_enable_single_replica_insert` spelling is registered there, so
`SET`, `SET GLOBAL`, and `SELECT @@...` now hit `ERR_UNKNOWN_SYSTEM_VARIABLE`.
Please add the compatibility names (or normalize the experimental prefix in the
removed-name lookup) and cover these legacy statements with a `VariableMgr`
test.
##########
be/src/load/channel/tablets_channel.cpp:
##########
@@ -418,8 +412,6 @@ Status TabletsChannel::close(LoadChannel* parent, const
PTabletWriterAddBlockReq
}
}
- _write_single_replica = req.write_single_replica();
Review Comment:
An old sink executor can still target this upgraded BE as the selected
tablet master. In that request, the old FE put only this master in `location`;
every other replica exists solely in EOS `req.slave_tablet_nodes`. With the
receiver-side handling removed, this method ignores `write_single_replica`,
commits only the local writer, and returns OK with no slave successes. The old
sender records one commit info and the FE fails quorum only after the transfer.
The `VTabletWriter::_init` guard does not run on this receiving master, so
please reject the legacy add-block request before commit (or translate it to
all listed replicas) and add an old-sender/new-master mixed-version test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
