csun5285 commented on code in PR #27764:
URL: https://github.com/apache/doris/pull/27764#discussion_r1413790948
##########
be/src/service/internal_service.cpp:
##########
@@ -852,6 +855,110 @@ void PInternalServiceImpl::_get_column_ids_by_tablet_ids(
response->mutable_status()->set_status_code(TStatusCode::OK);
}
+template <class RPCResponse>
+struct AsyncRPCContext {
+ RPCResponse response;
+ brpc::Controller cntl;
+ brpc::CallId cid;
+};
+
+void
PInternalServiceImpl::fetch_remote_tablet_schema(google::protobuf::RpcController*
controller,
+ const
PFetchRemoteSchemaRequest* request,
+
PFetchRemoteSchemaResponse* response,
+
google::protobuf::Closure* done) {
+ bool ret = _heavy_work_pool.try_offer([request, response, done]() {
+ brpc::ClosureGuard closure_guard(done);
+ Status st = Status::OK();
+ if (request->is_coordinator()) {
+ // Spawn rpc request to none coordinator nodes, and finally merge
them all
+ PFetchRemoteSchemaRequest remote_request(*request);
+ // set it none coordinator to get merged schema
+ remote_request.set_is_coordinator(false);
+ using PFetchRemoteTabletSchemaRpcContext =
AsyncRPCContext<PFetchRemoteSchemaResponse>;
+ std::vector<PFetchRemoteTabletSchemaRpcContext> rpc_contexts(
+ request->tablet_location_size());
+ for (int i = 0; i < request->tablet_location_size(); ++i) {
+ std::string host = request->tablet_location(i).host();
+ int32_t brpc_port = request->tablet_location(i).brpc_port();
+ std::shared_ptr<PBackendService_Stub> stub(
+
ExecEnv::GetInstance()->brpc_internal_client_cache()->get_client(
+ host, brpc_port));
+ rpc_contexts[i].cid = rpc_contexts[i].cntl.call_id();
+ stub->fetch_remote_tablet_schema(&rpc_contexts[i].cntl,
&remote_request,
+ &rpc_contexts[i].response,
brpc::DoNothing());
+ }
+ std::vector<TabletSchemaSPtr> schemas;
+ for (auto& rpc_context : rpc_contexts) {
+ brpc::Join(rpc_context.cid);
+ if (!st.ok()) {
+ // make sure all flying rpc request is joined
+ continue;
+ }
+ if (rpc_context.cntl.Failed()) {
+ LOG(WARNING) << "fetch_remote_tablet_schema rpc err:"
+ << rpc_context.cntl.ErrorText();
+
ExecEnv::GetInstance()->brpc_internal_client_cache()->erase(
+ rpc_context.cntl.remote_side());
+ st = Status::InternalError("fetch_remote_tablet_schema rpc
err: {}",
+ rpc_context.cntl.ErrorText());
+ }
+ if (rpc_context.response.status().status_code() != 0) {
+ st = Status::create(rpc_context.response.status());
+ }
+ if (rpc_context.response.has_merged_schema()) {
+ TabletSchemaSPtr schema = std::make_shared<TabletSchema>();
+ schema->init_from_pb(rpc_context.response.merged_schema());
+ schemas.push_back(schema);
+ }
+ }
+ if (!schemas.empty() && st.ok()) {
+ // merge all
+ TabletSchemaSPtr merged_schema =
+
vectorized::schema_util::get_least_common_schema(schemas, nullptr);
+ VLOG_DEBUG << "dump schema:" <<
merged_schema->dump_structure();
+ merged_schema->to_schema_pb(response->mutable_merged_schema());
+ }
+ st.to_protobuf(response->mutable_status());
+ return;
+ }
+
+ // This is not a coordinator, get it's tablet and merge schema
+ std::vector<int64_t> target_tablets;
+ for (int i = 0; i < request->tablet_location_size(); ++i) {
+ const auto& location = request->tablet_location(i);
+ auto backend = BackendOptions::get_local_backend();
+ // If this is the target backend
+ if (backend.host == location.host() && config::brpc_port ==
location.brpc_port()) {
+ target_tablets.assign(location.tablet_id().begin(),
location.tablet_id().end());
+ break;
+ }
+ }
+ if (!target_tablets.empty()) {
+ std::vector<TabletSchemaSPtr> tablet_schemas;
+ for (int64_t tablet_id : target_tablets) {
+ TabletSharedPtr tablet =
+
StorageEngine::instance()->tablet_manager()->get_tablet(tablet_id, false);
+ if (tablet == nullptr) {
+ // just ignore
Review Comment:
done
--
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: commits-unsubscr...@doris.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
