| 40 | {} |
| 41 | |
| 42 | void RemoteInserter::initialize() |
| 43 | { |
| 44 | ClientInfo modified_client_info = client_info; |
| 45 | modified_client_info.query_kind = ClientInfo::QueryKind::SECONDARY_QUERY; |
| 46 | |
| 47 | Settings settings = insert_settings; |
| 48 | /// With current protocol it is impossible to avoid deadlock in case of send_logs_level!=none. |
| 49 | /// |
| 50 | /// RemoteInserter send Data blocks/packets to the remote shard, |
| 51 | /// while remote side can send Log packets to the initiator (this RemoteInserter instance). |
| 52 | /// |
| 53 | /// But it is not enough to pull Log packets just before writing the next block |
| 54 | /// since there is no way to ensure that all Log packets had been consumed. |
| 55 | /// |
| 56 | /// And if enough Log packets will be queued by the remote side, |
| 57 | /// it will wait send_timeout until initiator will consume those packets, |
| 58 | /// while initiator already starts writing Data blocks, |
| 59 | /// and will not consume Log packets. |
| 60 | /// |
| 61 | /// So that is why send_logs_level had been disabled here. |
| 62 | settings[Setting::send_logs_level] = "none"; |
| 63 | /** Send query and receive "header", that describes table structure. |
| 64 | * Header is needed to know, what structure is required for blocks to be passed to 'write' method. |
| 65 | */ |
| 66 | /// TODO (vnemkov): figure out should we pass additional roles in this case or not. |
| 67 | connection.sendQuery( |
| 68 | timeouts, query, /* query_parameters */ {}, "", QueryProcessingStage::Complete, &settings, &modified_client_info, false, /* external_roles */ {}, {}); |
| 69 | |
| 70 | while (true) |
| 71 | { |
| 72 | Packet packet = connection.receivePacket(); |
| 73 | |
| 74 | if (Protocol::Server::Data == packet.type) |
| 75 | { |
| 76 | header = packet.block; |
| 77 | break; |
| 78 | } |
| 79 | if (Protocol::Server::Exception == packet.type) |
| 80 | { |
| 81 | packet.exception->rethrow(); |
| 82 | break; |
| 83 | } |
| 84 | if (Protocol::Server::Log == packet.type) |
| 85 | { |
| 86 | /// Pass logs from remote server to client |
| 87 | if (auto log_queue = CurrentThread::getInternalTextLogsQueue()) |
| 88 | log_queue->pushBlock(std::move(packet.block)); |
| 89 | } |
| 90 | else if (Protocol::Server::TableColumns == packet.type) |
| 91 | { |
| 92 | /// Server could attach ColumnsDescription in front of stream for column defaults. There's no need to pass it through cause |
| 93 | /// client's already got this information for remote table. Ignore. |
| 94 | } |
| 95 | else if (Protocol::Server::Progress == packet.type) |
| 96 | { |
| 97 | /// Progress packets are ignored |
| 98 | } |
| 99 | else |
nothing calls this directly
no test coverage detected