| 418 | } |
| 419 | |
| 420 | void RemoteQueryExecutor::sendQueryUnlocked(ClientInfo::QueryKind query_kind, AsyncCallback async_callback) |
| 421 | { |
| 422 | /// Emulate a concurrent cancel() landing right before the query is sent. |
| 423 | fiu_do_on(FailPoints::remote_query_executor_cancel_before_send, { was_cancelled = true; }); |
| 424 | |
| 425 | if (sent_query || was_cancelled) |
| 426 | return; |
| 427 | |
| 428 | connections = create_connections(async_callback); |
| 429 | AsyncCallbackSetter<IConnections> async_callback_setter(connections.get(), async_callback); |
| 430 | |
| 431 | const auto & settings = context->getSettingsRef(); |
| 432 | if (isReplicaUnavailable() || needToSkipUnavailableShard()) |
| 433 | { |
| 434 | /// To avoid sending the query again in the read(), we need to update the following flags: |
| 435 | was_cancelled = true; |
| 436 | finished = true; |
| 437 | sent_query = true; |
| 438 | |
| 439 | /// We need to tell the coordinator not to wait for this replica. |
| 440 | if (extension && extension->parallel_reading_coordinator) |
| 441 | { |
| 442 | chassert(extension->replica_info); |
| 443 | extension->parallel_reading_coordinator->markReplicaAsUnavailable(extension->replica_info->number_of_current_replica); |
| 444 | } |
| 445 | |
| 446 | return; |
| 447 | } |
| 448 | |
| 449 | established = true; |
| 450 | |
| 451 | auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(settings); |
| 452 | ClientInfo modified_client_info = context->getClientInfo(); |
| 453 | modified_client_info.query_kind = query_kind; |
| 454 | |
| 455 | if (extension) |
| 456 | modified_client_info.collaborate_with_initiator = true; |
| 457 | |
| 458 | // Collect all roles granted on this node and pass those to the remote node |
| 459 | Strings local_granted_roles; |
| 460 | if (context->getSettingsRef()[Setting::push_external_roles_in_interserver_queries]) |
| 461 | { |
| 462 | auto user = context->getAccessControl().read<User>(modified_client_info.initial_user, false); |
| 463 | boost::container::flat_set<String> granted_roles; |
| 464 | if (user) |
| 465 | { |
| 466 | const auto & access_control = context->getAccessControl(); |
| 467 | for (const auto & e : user->granted_roles.getElements()) |
| 468 | { |
| 469 | // `tryReadNames` instead of `readNames` because the original user might have a dropped role. |
| 470 | auto names = access_control.tryReadNames(e.ids); |
| 471 | granted_roles.insert(names.begin(), names.end()); |
| 472 | } |
| 473 | } |
| 474 | local_granted_roles.insert(local_granted_roles.end(), granted_roles.begin(), granted_roles.end()); |
| 475 | } |
| 476 | |
| 477 | if (distributed_fanout > 0) |
no test coverage detected