{

const Settings & settings = context->getSettingsRef();

ThrottlerPtr user_level_throttler;

if (auto process_list_element = context->getProcessListElement())

user_level_throttler = process_list_element->getUserNetworkThrottler();

/// Network bandwidth limit, if needed.

ThrottlerPtr throttler;

if (settings[Setting::max_network_bandwidth] || settings[Setting::max_network_bytes])

{

throttler = std::make_shared<Throttler>(

settings[Setting::max_network_bandwidth],

settings[Setting::max_network_bytes],

"Limit for bytes to send or receive over network exceeded.",

user_level_throttler);

}

else

throttler = user_level_throttler;

return throttler;

const String & query_,

SharedHeader header_,

ContextPtr context_,

const Scalars & scalars_,

const Tables & external_tables_,

QueryProcessingStage::Enum stage_,

std::shared_ptr<const QueryPlan> query_plan_,

std::optional<Extension> extension_,

GetPriorityForLoadBalancing::Func priority_func_)

: header(header_)

, query(query_)

, query_plan(std::move(query_plan_))

, context(context_)

, scalars(scalars_)

, external_tables(external_tables_)

, stage(stage_)

, extension(extension_)

, skip_unavailable_shards(context->getSettingsRef()[Setting::skip_unavailable_shards])

, skip_unavailable_shards_mode(context->getSettingsRef()[Setting::skip_unavailable_shards_mode])

, priority_func(priority_func_)

, read_packet_type_separately(context->canUseParallelReplicasOnInitiator() && !context->getSettingsRef()[Setting::use_hedged_requests])

{

if (stage == QueryProcessingStage::QueryPlan && !query_plan)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Query plan is not passed for QueryPlan processing stage");

ConnectionPoolPtr pool,

const String & query_,

SharedHeader header_,

ContextPtr context_,

ThrottlerPtr throttler,

const Scalars & scalars_,

const Tables & external_tables_,

QueryProcessingStage::Enum stage_,

std::optional<Extension> extension_,

ConnectionPoolWithFailoverPtr connection_pool_with_failover_,

std::shared_ptr<const QueryPlan> query_plan_)

: RemoteQueryExecutor(query_, header_, context_, scalars_, external_tables_, stage_, query_plan_, extension_)

{

create_connections = [this, pool, throttler, extension_, connection_pool_with_failover_](AsyncCallback)

{

const Settings & settings = context->getSettingsRef();

auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithoutFailover(settings)

.withUnsecureConnectionTimeout(settings[Setting::parallel_replicas_connect_timeout_ms])

.withSecureConnectionTimeout(settings[Setting::parallel_replicas_connect_timeout_ms]);

ConnectionPoolWithFailover::TryResult result;

std::string fail_message;

if (main_table)

{

auto table_name = main_table.getQualifiedName();

ConnectionEstablisher connection_establisher(pool, &timeouts, settings, log, &table_name);

connection_establisher.run(result, fail_message);

}

else

{

ConnectionEstablisher connection_establisher(pool, &timeouts, settings, log, nullptr);

connection_establisher.run(result, fail_message);

}

ConnectionPoolEntries connection_entries;

if (!result.entry.isNull() && result.is_usable)

{

chassert(result.entry->isConnected());

const auto protocol_version = result.entry->getServerRevision(ConnectionTimeouts{});

const auto parallel_replicas_version = result.entry->getParallelReplicasProtocolVersion();

const auto query_plan_serialization_version = result.entry->getQueryPlanSerializationVersion();

if (extension_ && extension_->parallel_reading_coordinator)

{

// consider only replicas with support of stream id, otherwise we can get incorrect result

// replicas with older version considered as unavailable

if (protocol_version >= DBMS_MIN_REVISION_WITH_PARALLEL_REPLICAS

&& parallel_replicas_version >= DBMS_PARALLEL_REPLICAS_MIN_VERSION_WITH_STREAM_ID

&& (!query_plan || query_plan_serialization_version >= DBMS_MIN_QUERY_PLAN_SERIALIZATION_VERSION_WITH_PARALLEL_REPLICAS))

{

ProfileEvents::increment(ProfileEvents::ParallelReplicasAvailableCount);

connection_entries.emplace_back(std::move(result.entry));

}

else

{

LOG_DEBUG(

log,

"Disconnecting replica {} (protocol_version={}, parallel_replicas_version={}, "

"query_plan_serialization_version={}): "

"remote replica doesn't support stream id (requires parallel_replicas_version >= {}) or query plan serialization "

"for parallel replicas (requires query_plan_serialization_version >= {})",

result.entry->getDescription(),

protocol_version,

parallel_replicas_version,

query_plan_serialization_version,

DBMS_PARALLEL_REPLICAS_MIN_VERSION_WITH_STREAM_ID,

DBMS_MIN_QUERY_PLAN_SERIALIZATION_VERSION_WITH_PARALLEL_REPLICAS);

result.entry->disconnect();

}

}

else

{

connection_entries.emplace_back(std::move(result.entry));

}

}

else

{

chassert(!fail_message.empty());

if (result.entry.isNull())

{

LOG_DEBUG(log, "Failed to connect to replica {}. {}", pool->getAddress(), fail_message);

if (connection_pool_with_failover_)

connection_pool_with_failover_->incrementErrorCount(pool);

}

else

LOG_DEBUG(log, "Replica is not usable for remote query execution: {}. {}", pool->getAddress(), fail_message);

}

auto res = std::make_unique<MultiplexedConnections>(std::move(connection_entries), context, throttler);

if (extension_ && extension_->replica_info)

res->setReplicaInfo(*extension_->replica_info);

return res;

};

Connection & connection,

const String & query_,

SharedHeader header_,

ContextPtr context_,

ThrottlerPtr throttler,

const Scalars & scalars_,

const Tables & external_tables_,

QueryProcessingStage::Enum stage_,

std::optional<Extension> extension_)

: RemoteQueryExecutor(query_, header_, context_, scalars_, external_tables_, stage_, nullptr, extension_)

{

create_connections = [this, &connection, throttler, extension_](AsyncCallback)

{

auto res = std::make_unique<MultiplexedConnections>(connection, context, throttler);

if (extension_ && extension_->replica_info)

res->setReplicaInfo(*extension_->replica_info);

return res;

};

ConnectionPoolEntries && connections_,

const String & query_,

SharedHeader header_,

ContextPtr context_,

const ThrottlerPtr & throttler,

const Scalars & scalars_,

const Tables & external_tables_,

QueryProcessingStage::Enum stage_,

std::shared_ptr<const QueryPlan> query_plan_,

std::optional<Extension> extension_,

ConnectionPoolWithFailoverPtr pool)

: RemoteQueryExecutor(query_, header_, context_, scalars_, external_tables_, stage_, std::move(query_plan_), extension_)

{

/// Capture `pool` in the lambda to prevent the connection pool from being destroyed

/// while entries are still in use. The Entry objects hold raw references (via PoolEntryHelper)

/// back to the pool's internal PooledObject and PoolBase structures, so the pool must

/// outlive all Entry objects.

create_connections = [this, connections_, throttler, extension_, pool](AsyncCallback) mutable

{

auto res = std::make_unique<MultiplexedConnections>(std::move(connections_), context, throttler);

if (extension_ && extension_->replica_info)

res->setReplicaInfo(*extension_->replica_info);

return res;

};

const ConnectionPoolWithFailoverPtr & pool,

const String & query_,

SharedHeader header_,

ContextPtr context_,

const ThrottlerPtr & throttler,

const Scalars & scalars_,

const Tables & external_tables_,

QueryProcessingStage::Enum stage_,

std::shared_ptr<const QueryPlan> query_plan_,

std::optional<Extension> extension_,

GetPriorityForLoadBalancing::Func priority_func_)

: RemoteQueryExecutor(query_, header_, context_, scalars_, external_tables_, stage_, std::move(query_plan_), extension_, priority_func_)

{

create_connections = [this, pool, throttler](AsyncCallback async_callback)->std::unique_ptr<IConnections>

{

const Settings & current_settings = context->getSettingsRef();

auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(current_settings);

#if defined(OS_LINUX)

if (current_settings[Setting::use_hedged_requests])

{

std::shared_ptr<QualifiedTableName> table_to_check = nullptr;

if (main_table)

table_to_check = std::make_shared<QualifiedTableName>(main_table.getQualifiedName());

auto res = std::make_unique<HedgedConnections>(

pool, context, timeouts, throttler, pool_mode, table_to_check, std::move(async_callback), priority_func);

if (extension && extension->replica_info)

res->setReplicaInfo(*extension->replica_info);

return res;

}

#endif

ConnectionPoolEntries connection_entries;

std::optional<bool> skip_unavailable_endpoints;

if (extension && extension->parallel_reading_coordinator)

skip_unavailable_endpoints = true;

if (main_table)

{

auto try_results = pool->getManyChecked(

timeouts,

current_settings,

pool_mode,

main_table.getQualifiedName(),

std::move(async_callback),

skip_unavailable_endpoints,

priority_func);

connection_entries.reserve(try_results.size());

for (auto & try_result : try_results)

connection_entries.emplace_back(std::move(try_result.entry));

}

else

{

connection_entries = pool->getMany(

timeouts, current_settings, pool_mode, std::move(async_callback), skip_unavailable_endpoints, priority_func);

}

auto res = std::make_unique<MultiplexedConnections>(std::move(connection_entries), context, throttler);

if (extension && extension->replica_info)

res->setReplicaInfo(*extension->replica_info);

return res;

};

{

/// Special case when reader doesn't care about result structure. Deprecated and used only in Benchmark, PerformanceTest.

if (header.empty())

return block;

Block res;

res.info = block.info;

for (const auto & elem : header)

{

ColumnPtr column;

if (elem.column && isColumnConst(*elem.column))

{

/// We expect constant column in block.

/// If block is not empty, then get value for constant from it,

/// because it may be different for remote server for functions like version(), uptime(), ...

if (block.rows() > 0 && block.has(elem.name))

{

/// Const column is passed as materialized. Get first value from it.

///

/// TODO: check that column contains the same value.

/// TODO: serialize const columns.

auto col = block.getByName(elem.name);

if (const auto * blob = typeid_cast<const ColumnBLOB *>(col.column.get()))

col.column = blob->convertFrom();

col.column = col.column->cut(0, 1);

column = castColumn(col, elem.type);

if (!isColumnConst(*column))

column = ColumnConst::create(column, block.rows());

else

/// It is not possible now. Just in case we support const columns serialization.

column = column->cloneResized(block.rows());

}

else

column = elem.column->cloneResized(block.rows());

}

else

{

const auto & col = block.getByName(elem.name);

if (auto * blob = typeid_cast<ColumnBLOB *>(col.column->assumeMutable().get()))

{

blob->addCast(col.type, elem.type);

column = col.column;

}

else

column = castColumn(col, elem.type);

}

res.insert({column, elem.type, elem.name});

}

return res;

{

/// Query cannot be canceled in the middle of the send query,

/// since there are multiple packets:

/// - Query

/// - Data (multiple times)

///

/// And after the Cancel packet none Data packet can be sent, otherwise the remote side will throw:

///

/// Unexpected packet Data received from client

///

LockAndBlocker guard(was_cancelled_mutex);

sendQueryUnlocked(query_kind, async_callback);

{

/// Emulate a concurrent cancel() landing right before the query is sent.

fiu_do_on(FailPoints::remote_query_executor_cancel_before_send, { was_cancelled = true; });

if (sent_query || was_cancelled)

return;

connections = create_connections(async_callback);

AsyncCallbackSetter<IConnections> async_callback_setter(connections.get(), async_callback);

const auto & settings = context->getSettingsRef();

if (isReplicaUnavailable() || needToSkipUnavailableShard())

{

/// To avoid sending the query again in the read(), we need to update the following flags:

was_cancelled = true;

finished = true;

sent_query = true;

/// We need to tell the coordinator not to wait for this replica.

if (extension && extension->parallel_reading_coordinator)

{

chassert(extension->replica_info);

extension->parallel_reading_coordinator->markReplicaAsUnavailable(extension->replica_info->number_of_current_replica);

}

return;

}

established = true;

auto timeouts = ConnectionTimeouts::getTCPTimeoutsWithFailover(settings);

ClientInfo modified_client_info = context->getClientInfo();

modified_client_info.query_kind = query_kind;

/// A distributed query must carry a known initiator version: the receiving server uses it for

/// version-gated compatibility decisions (e.g. whether to enable the analyzer, see `TCPHandler`).

/// A zero version means the initiating query context was not populated as an initial query

/// (a real client always reports its version, and a server that (re-)initiates a query fills it

/// with its own version). Sending zero silently triggers wrong compatibility downgrades on the

/// remote, so fail loudly instead.

if (modified_client_info.client_version_major == 0

&& modified_client_info.client_version_minor == 0

&& modified_client_info.client_version_patch == 0)

throw Exception(ErrorCodes::LOGICAL_ERROR,

"Sending a distributed query with unknown (zero) client version. "

"The query context was not initialized as an initial query");

/// Forward this node's current roles so the remote scopes row policies the same way (gated by the setting).

/// Reset first against stale/injected values, and skip when initial_user was rewritten (remote(user=>...)).

modified_client_info.current_roles.reset();

if (context->getSettingsRef()[Setting::push_external_roles_in_interserver_queries]

&& modified_client_info.initial_user == modified_client_info.current_user)

{

const auto & access_control = context->getAccessControl();

Strings current_role_names;

for (const auto & role_id : context->getCurrentRoles())

{

/// tryReadName: skip a concurrently-dropped role (its policies already target nobody).

if (auto name = access_control.tryReadName(role_id))

current_role_names.push_back(*name);

}

modified_client_info.current_roles = std::move(current_role_names);

}

if (extension)

modified_client_info.collaborate_with_initiator = true;

// Collect all roles granted on this node and pass those to the remote node

Strings local_granted_roles;

if (context->getSettingsRef()[Setting::push_external_roles_in_interserver_queries])

{

auto user = context->getAccessControl().read<User>(modified_client_info.initial_user, false);

boost::container::flat_set<String> granted_roles;

if (user)

{

const auto & access_control = context->getAccessControl();

for (const auto & e : user->granted_roles.getElements())

{

// `tryReadNames` instead of `readNames` because the original user might have a dropped role.

auto names = access_control.tryReadNames(e.ids);

granted_roles.insert(names.begin(), names.end());

}

}

local_granted_roles.insert(local_granted_roles.end(), granted_roles.begin(), granted_roles.end());

}

if (distributed_fanout > 0)

connections->setDistributedFanout(distributed_fanout);

connections->sendQuery(timeouts, query, query_id, stage, modified_client_info, true, local_granted_roles);

established = false;

sent_query = true;

sendScalars();

if (query_plan)

connections->sendQueryPlan(*query_plan);

sendExternalTables();

{

#if defined(OS_LINUX) || defined(OS_DARWIN)

LockAndBlocker lock(was_cancelled_mutex);

if (was_cancelled)

return -1;

if (!read_context)

read_context = std::make_unique<ReadContext>(

*this,

/*suspend_when_query_sent*/ true,

read_packet_type_separately);

/// If query already sent, do nothing. Note that we cannot use sent_query flag here,

/// because we can still be in process of sending scalars or external tables.

if (read_context->isQuerySent())

return -1;

read_context->resume();

if (read_context->isQuerySent())

return -1;

ProfileEvents::increment(ProfileEvents::SuspendSendingQueryToShard); /// Mostly for testing purposes.

return read_context->getFileDescriptor();

#else

sendQuery();

return -1;

#endif

{

while (true)

{

auto res = read();

if (res.getType() == ReadResult::Type::Data)

return res.getBlock();

}

{

if (!sent_query)

{

sendQuery();

/// `connections` stays null if sendQuery() was cancelled before sending,

/// so guard the dereference below (as every other use of it does).

{

LockAndBlocker lock(was_cancelled_mutex);

if (was_cancelled)

return ReadResult(Block());

}

if (context->getSettingsRef()[Setting::skip_unavailable_shards] && (0 == connections->size()))

return ReadResult(Block());

}

while (true)

{

{

LockAndBlocker lock(was_cancelled_mutex);

if (was_cancelled)

return ReadResult(Block());

}

/// Parks the reader in the window this fix is about: `was_cancelled` has just been checked

/// and the mutex released, so a parallel `onUpdatePorts` can cancel and drain these

/// connections before `receivePacket` below runs.

fiu_do_on(FailPoints::remote_query_executor_receive_packet_pause, {

in_receive_packet_window = true;

FailPointInjection::notifyPauseAndWaitForResume(FailPoints::remote_query_executor_receive_packet_pause);

in_receive_packet_window = false;

});

auto packet = connections->receivePacket();

LockAndBlocker lock(was_cancelled_mutex);

if (was_cancelled)

return ReadResult(Block());

auto anything = processPacket(std::move(packet));

if (anything.getType() == ReadResult::Type::Data || anything.getType() == ReadResult::Type::ParallelReplicasToken)

return anything;

}

{

#if defined(OS_LINUX) || defined(OS_DARWIN)

if (!read_context)

{

LockAndBlocker lock(was_cancelled_mutex);

if (was_cancelled)

return ReadResult(Block());

read_context = std::make_unique<ReadContext>(

*this,

/*suspend_when_query_sent*/ false,

read_packet_type_separately);

}

while (true)

{

LockAndBlocker lock(was_cancelled_mutex);

if (was_cancelled)

return ReadResult(Block());

if (packet_in_progress)

{

chassert(read_context->readPacketTypeSeparately());

chassert(read_context->hasReadTillPacketType());

/// packet type is handled already, read and parse packet itself

if (!read_context->hasReadPacket() && !read_context->read())

return ReadResult(read_context->getFileDescriptor());

packet_in_progress = false;

auto read_result = processPacket(read_context->getPacket());

if (read_result.getType() == ReadResult::Type::Data || read_result.getType() == ReadResult::Type::ParallelReplicasToken)

return read_result;

}

read_context->resume();

if (isReplicaUnavailable() || needToSkipUnavailableShard())

{

/// We need to tell the coordinator not to wait for this replica.

/// But at this point it may lead to an incomplete result set, because

/// this replica committed to read some part of there data and then died.

if (extension && extension->parallel_reading_coordinator)

{

chassert(extension->parallel_reading_coordinator);

extension->parallel_reading_coordinator->markReplicaAsUnavailable(extension->replica_info->number_of_current_replica);

}

return ReadResult(Block());

}

/// Check if packet is not ready yet.

if (read_context->isInProgress())

return ReadResult(read_context->getFileDescriptor());

/// if reading separately packet header and body enabled, try to read packet itself this time

if (read_context->readPacketTypeSeparately() && !read_context->hasReadPacket() && !read_context->read())

return ReadResult(read_context->getFileDescriptor());

auto read_result = processPacket(read_context->getPacket());

if (read_result.getType() == ReadResult::Type::Data || read_result.getType() == ReadResult::Type::ParallelReplicasToken)

return read_result;

}

#else

return read();

#endif

{

switch (packet.type)

{

case Protocol::Server::MergeTreeReadTaskRequest:

chassert(packet.request.has_value());

processMergeTreeReadTaskRequest(packet.request.value());

return ReadResult(ReadResult::Type::ParallelReplicasToken);

case Protocol::Server::MergeTreeAllRangesAnnouncement:

chassert(packet.announcement.has_value());

processMergeTreeInitialReadAnnouncement(packet.announcement.value());

return ReadResult(ReadResult::Type::ParallelReplicasToken);

case Protocol::Server::ReadTaskRequest:

processReadTaskRequest();

break;

case Protocol::Server::PartUUIDs:

LOG_WARNING(

log,

"The remote server has sent no longer supported packet (Server::PartUUIDs). allow_experimental_query_deduplication feature "

"has been deprecated. Consider upgrading the remote server ({})",

connections->dumpAddresses());

break;

case Protocol::Server::Data:

/// Note: `packet.block.rows() > 0` means it's a header block.

/// We can actually return it, and the first call to RemoteQueryExecutor::read

/// will return earlier. We should consider doing it.

if (!packet.block.empty() && (packet.block.rows() > 0))

{

got_data_from_replica = true;

return ReadResult(adaptBlockStructure(packet.block, *header));

}

break; /// If the block is empty - we will receive other packets before EndOfStream.

case Protocol::Server::Exception:

got_exception_from_replica = true;

if (shouldIgnoreShardException(packet.exception->code()))

{

if (log)

LOG_ERROR(log,

"Ignoring exception from connection(s) {} due to `skip_unavailable_shards_mode` setting: {}",

connections->dumpAddresses(),

packet.exception->displayText());

reportShardSkipped();

/// The server terminated the query with this exception and will not send `EndOfStream`,

/// so mark the executor finished to signal end of data.

finished = true;

return ReadResult(Block{});

}

packet.exception->rethrow();

break;

case Protocol::Server::EndOfStream:

if (!connections->hasActiveConnections())

{

finished = true;

/// TODO: Replace with Type::Finished

return ReadResult(Block{});

}

break;

case Protocol::Server::Progress:

/** We use the progress from a remote server.

if (progress_callback)

progress_callback(packet.progress);

break;

case Protocol::Server::ProfileInfo:

/// Use own (client-side) info about read bytes, it is more correct info than server-side one.

if (profile_info_callback)

profile_info_callback(packet.profile_info);

break;

case Protocol::Server::Totals:

totals = packet.block;

if (!totals.empty())

totals = adaptBlockStructure(totals, *header);

break;

case Protocol::Server::Extremes:

extremes = packet.block;

if (!extremes.empty())

extremes = adaptBlockStructure(packet.block, *header);

break;

case Protocol::Server::Log:

/// Pass logs from remote server to client

if (auto log_queue = CurrentThread::getInternalTextLogsQueue())

log_queue->pushBlock(std::move(packet.block));

break;

case Protocol::Server::ProfileEvents:

/// Pass profile events from remote server to client

if (auto profile_queue = CurrentThread::getInternalProfileEventsQueue())

if (!profile_queue->emplace(std::move(packet.block)))

throw Exception(ErrorCodes::SYSTEM_ERROR, "Could not push into profile queue");

break;

case Protocol::Server::TimezoneUpdate:

break;

default:

got_unknown_packet_from_replica = true;

throw Exception(

ErrorCodes::UNKNOWN_PACKET_FROM_SERVER,

"Unknown packet {} from one of the following replicas: {}",

packet.type,

connections->dumpAddresses());

}

return ReadResult(ReadResult::Type::Nothing);

{

/// A ReadTaskRequest arrives only from a worker running a cluster table function or object storage

/// source with distributed reads. Serving it needs a task iterator, which only the legitimate

/// dispatch paths install; its absence means the source was reached through an outer distribution.

if (!extension)

throw Exception(

ErrorCodes::BAD_ARGUMENTS,

"A cluster table function (s3Cluster, urlCluster, fileCluster, ...) cannot be nested inside "

"another distributed query");

if (!extension->task_iterator)

{

if (extension->parallel_reading_coordinator)

throw Exception(

ErrorCodes::BAD_ARGUMENTS,

"A cluster table function or object storage cluster source cannot use distributed "

"processing inside a query that runs with parallel replicas");

throw Exception(

ErrorCodes::LOGICAL_ERROR,

"Received a cluster function read task request, but the query executor has neither a task "

"iterator nor a parallel replicas coordinator");

}

if (!extension->replica_info)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Replica info is not initialized");

ProfileEvents::increment(ProfileEvents::ReadTaskRequestsReceived);

auto response = (*extension->task_iterator)(extension->replica_info->number_of_current_replica);

connections->sendClusterFunctionReadTaskResponse(*response);

{

if (!extension || !extension->parallel_reading_coordinator)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Coordinator for parallel reading from replicas is not initialized");

ProfileEvents::increment(ProfileEvents::MergeTreeReadTaskRequestsReceived);

auto response = extension->parallel_reading_coordinator->handleRequest(std::move(request));

connections->sendMergeTreeReadTaskResponse(response);

{

if (!extension || !extension->parallel_reading_coordinator)

throw Exception(ErrorCodes::LOGICAL_ERROR, "Coordinator for parallel reading from replicas is not initialized");

/// Followers only block on the response when they actually need it (in-order modes pin

/// per-split parts via the response). In `Default` mode the caller discards the response,

/// so the round-trip would be pure overhead — skip it on both sides.

const bool send_response = announcement.mode != CoordinationMode::Default;

announcement_received = true;

auto response = extension->parallel_reading_coordinator->handleInitialAllRangesAnnouncement(std::move(announcement));

if (send_response)

connections->sendMergeTreeAllRangesAnnouncementResponse(response);

{

LockAndBlocker guard(was_cancelled_mutex);

/** If one of:

if (!isQueryPending() || hasThrownException() || was_cancelled)

{

/// If the query was never sent there is nothing to drain, but we must still mark the

/// executor as finished. Otherwise a RemoteSource whose output is closed before it sends

/// its query (e.g. an empty-build ANY INNER JOIN that short-circuits the probe side) keeps

/// re-entering its drain path via prepare()/work() and spins forever, because isFinished()

/// never becomes true.

if (!sent_query)

{

/// Also mark the executor cancelled, not just finished. `RemoteSource::work()` may

/// already be queued for execution (its `prepare()` ran before the output port was

/// closed), and both `sendQuery` and `sendQueryAsync` gate only on `was_cancelled` -

/// never on `finished`. Without this the query is still sent after we declared the

/// executor finished, and nothing releases it afterwards: `finish()` returns early

/// from here on, and the destructor's `isQueryPending()` is false because `finished`

/// is set, so the connection is returned to the pool without a `Cancel` packet and

/// without a disconnect. A parallel-replicas follower is then left blocked in

/// `receivePartitionMergeTreeReadTaskResponse` for the whole `receive_timeout`,

/// holding the table's shared lock and stalling a subsequent `DROP TABLE` (#109265).

was_cancelled = true;

finished = true;

}

else if (was_cancelled && !finished && connections)

{

/// The query was already cancelled (e.g. concurrently from the pipeline) but its

/// connections may still hold undelivered packets - the server keeps sending the data,

/// `ProfileInfo` and `EndOfStream` that were produced before it observed the cancel.

/// We do not drain them here after cancellation, because the read side may already be

/// torn down and reading from it could throw or crash (see #95466). But such connections

/// must not be returned to the connection pool in this out-of-sync state - otherwise the

/// next user of the connection would read a stale packet during establishment, failing

/// with "Unexpected packet from server (expected TablesStatusResponse, got ProfileInfo)"

/// (see #93018). So disconnect them, forcing a clean reconnect on reuse. This mirrors the

/// cleanup done in the destructor, but performs it eagerly so it cannot be skipped if

/// `finished` later becomes true through another path.

connections->disconnect();

finished = true;

}

return;

}

/// To make sure finish is only called once

SCOPE_EXIT({ finished = true; });

/** If you have not read all the data yet, but they are no longer needed.

/// Send the request to abort the execution of the request, if not already sent.

tryCancel("Cancelling query because enough data has been read");

/// If connections weren't created yet, query wasn't sent or was already finished, nothing to do.

if (!connections || !sent_query || finished)

return;

/// `tryCancel` above may have torn the read side down without draining the packet that was in

/// flight. The loop below reads from those same connections, and reading from a canceled buffer

/// aborts ("ReadBuffer is canceled. Can't read from it."). Disconnect instead of draining, exactly

/// as the already-cancelled branch above does and for the same reasons: the read side may be gone

/// (#95466), and an out-of-sync connection must not go back to the pool (#93018). `SCOPE_EXIT`

/// marks the executor finished on the way out.

if (drain_was_skipped)

{

connections->disconnect();

return;

}

/// Get the remaining packets so that there is no out of sync in the connections to the replicas.

/// We do this manually instead of calling drain() because we want to process Log, ProfileEvents and Progress

/// packets that had been sent before the connection is fully finished in order to have final statistics of what

/// was executed in the remote queries

while (connections->hasActiveConnections() && !finished)

{

Packet packet = connections->receivePacket();

switch (packet.type)