**Describe what's wrong**
some requests do not respect the limits set by the max_execution_time setting and are executed for an arbitrarily long time.
this has been seen with queries containing the expressions `global in (_subquery)` or `global join _data`.
the trace_log for these queries contains calls inside the `DB::Context::initializeexternaltablesifset()`
it is expected that during the execution of the request, the server will check from time to time whether it has gone beyond the limits.

**How to reproduce**

```
┌─version()──┐
│ 21.8.10.19 │
└────────────┘
```
clickhouse-client, http
execute query `select column, count() from table1 where col global in (select col from table2 where cond) group by column` with distributed tables table1/table2 on sufficiently large cluster.

**Expected behavior**

after the timeout expires, the query ends with an error of 159/160/209

**Actual behavior**
remote servers spend unlimited time executing their part of an external distributed request. trace_log of the corresponding requests looks like this: 
```
/usr/bin/clickhouse	DB::Block::~Block()
/usr/bin/clickhouse	std::__1::__shared_ptr_pointer<std::__1::vector<DB::Block, std::__1::allocator<DB::Block> > const*, std::__1::default_delete<std::__1::vector<DB::Block, std::__1::allocator<DB::Block> > const>, std::__1::allocator<std::__1::vector<DB::Block, std::__1::allocator<DB::Block> > const> >::__on_zero_shared()
/usr/bin/clickhouse	DB::MemoryBlockOutputStream::writeSuffix()
/usr/bin/clickhouse	DB::TCPHandler::receiveData(bool)
/usr/bin/clickhouse	DB::TCPHandler::receivePacket()
/usr/bin/clickhouse	DB::TCPHandler::readDataNext(unsigned long, long)
/usr/bin/clickhouse	
/usr/bin/clickhouse	DB::Context::initializeExternalTablesIfSet()
/usr/bin/clickhouse	
/usr/bin/clickhouse	DB::executeQuery(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::shared_ptr<DB::Context>, bool, DB::QueryProcessingStage::Enum, bool)
/usr/bin/clickhouse	DB::TCPHandler::runImpl()
/usr/bin/clickhouse	DB::TCPHandler::run()
/usr/bin/clickhouse	Poco::Net::TCPServerConnection::start()
/usr/bin/clickhouse	Poco::Net::TCPServerDispatcher::run()
/usr/bin/clickhouse	Poco::PooledThread::run()
/usr/bin/clickhouse	Poco::ThreadImpl::runnableEntry(void*)
/lib/x86_64-linux-gnu/libpthread-2.19.so	start_thread
/lib/x86_64-linux-gnu/libc-2.19.so	clone

/usr/bin/clickhouse	operator new(unsigned long)
/usr/bin/clickhouse	std::__1::__hash_table<std::__1::__hash_value_type<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, unsigned long>, std::__1::__unordered_map_hasher<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, std::__1::__hash_value_type<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, unsigned long>, std::__1::hash<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, std::__1::equal_to<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, true>, std::__1::__unordered_map_equal<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, std::__1::__hash_value_type<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, unsigned long>, std::__1::equal_to<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, std::__1::hash<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > >, true>, std::__1::allocator<std::__1::__hash_value_type<std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> >, unsigned long> > >::__rehash(unsigned long)
/usr/bin/clickhouse	void std::__1::allocator<DB::Block>::construct<DB::Block, DB::Block&>(DB::Block*, DB::Block&)
/usr/bin/clickhouse	DB::MemoryBlockOutputStream::writeSuffix()
/usr/bin/clickhouse	DB::TCPHandler::receiveData(bool)
/usr/bin/clickhouse	DB::TCPHandler::receivePacket()
/usr/bin/clickhouse	DB::TCPHandler::readDataNext(unsigned long, long)
/usr/bin/clickhouse	
/usr/bin/clickhouse	DB::Context::initializeExternalTablesIfSet()
/usr/bin/clickhouse	
/usr/bin/clickhouse	DB::executeQuery(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::shared_ptr<DB::Context>, bool, DB::QueryProcessingStage::Enum, bool)
/usr/bin/clickhouse	DB::TCPHandler::runImpl()
/usr/bin/clickhouse	DB::TCPHandler::run()
/usr/bin/clickhouse	Poco::Net::TCPServerConnection::start()
/usr/bin/clickhouse	Poco::Net::TCPServerDispatcher::run()
/usr/bin/clickhouse	Poco::PooledThread::run()
/usr/bin/clickhouse	Poco::ThreadImpl::runnableEntry(void*)
/lib/x86_64-linux-gnu/libpthread-2.19.so	start_thread
/lib/x86_64-linux-gnu/libc-2.19.so	clone

/usr/bin/clickhouse	std::__1::vector<DB::ColumnWithTypeAndName, std::__1::allocator<DB::ColumnWithTypeAndName> >::vector(std::__1::vector<DB::ColumnWithTypeAndName, std::__1::allocator<DB::ColumnWithTypeAndName> > const&)
/usr/bin/clickhouse	void std::__1::allocator<DB::Block>::construct<DB::Block, DB::Block&>(DB::Block*, DB::Block&)
/usr/bin/clickhouse	DB::MemoryBlockOutputStream::writeSuffix()
/usr/bin/clickhouse	DB::TCPHandler::receiveData(bool)
/usr/bin/clickhouse	DB::TCPHandler::receivePacket()
/usr/bin/clickhouse	DB::TCPHandler::readDataNext(unsigned long, long)
/usr/bin/clickhouse	
/usr/bin/clickhouse	DB::Context::initializeExternalTablesIfSet()
/usr/bin/clickhouse	
/usr/bin/clickhouse	DB::executeQuery(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, std::__1::shared_ptr<DB::Context>, bool, DB::QueryProcessingStage::Enum, bool)
/usr/bin/clickhouse	DB::TCPHandler::runImpl()
/usr/bin/clickhouse	DB::TCPHandler::run()
/usr/bin/clickhouse	Poco::Net::TCPServerConnection::start()
/usr/bin/clickhouse	Poco::Net::TCPServerDispatcher::run()
/usr/bin/clickhouse	Poco::PooledThread::run()
/usr/bin/clickhouse	Poco::ThreadImpl::runnableEntry(void*)
/lib/x86_64-linux-gnu/libpthread-2.19.so	start_thread
/lib/x86_64-linux-gnu/libc-2.19.so	clone
```

**Additional context**
additional heavy filters (like substring from long string column) in `cond` not only reduce the number of rows returned from the subquery, but also increase the overall query execution time.


{
    
    std::lock_guard guard(was_cancelled_mutex);
    if (was_cancelled)
        return;
    was_cancelled = true;
    if (read_context && *read_context)
    {
        
        
        
        
        
        
        
        
        (*read_context)->setTimer();
        (*read_context)->cancel();
    }
    connections->sendCancel();
    if (log)
        LOG_TRACE(log, "({}) {}", connections->dumpAddresses(), reason);
}
bool RemoteQueryExecutor::isQueryPending() const