 CREATE TABLE ladnl.dnl_log
(
    `event_date` Date,
    `date_id` UInt64,
    `tm` FixedString(8),
    `dt` DateTime DEFAULT toDateTime(concat(toString(event_date), ' ', tm)),
    `host_id` UInt32,
    `ip` Int32,
    `cc` FixedString(2),
    `code` UInt16,
    `topdir` String,
    `file` String,
    `size` UInt32,
    `flag` String,
    `uagent` String,
    `ok` UInt8,
    `appid` UInt16,
    `ver` UInt16 DEFAULT CAST(0, 'UInt16'),
    `ltype` UInt16 DEFAULT CAST(0, 'UInt16'),
    `keynum` Int16 DEFAULT CAST(0, 'Int16'),
    `md5` String,
    `serial` String,
    `pkid` UInt32,
    `pk_member` UInt16,
    `pk_product` UInt32,
    `cids` String,
    `sid` UInt16,
    `build` String,
    `sku` String,
    `iscommercial` UInt8,
    `istrial` UInt8,
    `isbeta` UInt8,
    `iscorporate` UInt8,
    `blocked` UInt8,
    `expdate` Date,
    `pcode` UInt16,
    `ssl` UInt8 DEFAULT 0,
    `ssl_reuse` UInt8 DEFAULT 0,
    `ip_num` String DEFAULT IPv4NumToString(toUInt32(ip)),
    `rport` UInt16 DEFAULT CAST(0, 'UInt16'),
    `tcp` String,
    `base` String,
    `comp` String
)
ENGINE = Distributed(dnl2019, 'ladnl', 'dnl_log_local', rand())


SELECT count(),sum(vol) FROM (SELECT count() as cnt,sum(size) as vol FROM ladnl.dnl_log WHERE event_date='2021-04-20' and serial global in (select md5 FROM ladnl.dnl_log WHERE event_date='2021-04-20' AND host_id=191 AND appid=2143 AND keynum>0 group by md5) group by serial HAVING count()=73)

2021.04.21 05:07:53.641943 [ 586494 ] {04d19dd0-692d-4dab-ab46-ac75cdc87eb8} <Error> DynamicQueryHandler: Code: 49, e.displayText() = DB::Exception: Can't initialize pipeline with empty pipe.: While executing Remote, Stack trace (when copying this message, always include the lines below):

0. DB::Exception::Exception(std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&, int, bool) @ 0x8914f0a in /usr/bin/clickhouse
1. DB::QueryPipeline::init(DB::Pipe) @ 0x101b03f1 in /usr/bin/clickhouse
2. DB::Connection::sendExternalTablesData(std::__1::vector<std::__1::unique_ptr<DB::ExternalTableData, std::__1::default_delete<DB::ExternalTableData> >, std::__1::allocator<std::__1::unique_ptr<DB::ExternalTableData, std::__1::default_delete<DB::ExternalTableData> > > >&) @ 0x1008c5db in /usr/bin/clickhouse
3. DB::HedgedConnections::processNewReplicaState(DB::HedgedConnectionsFactory::State, DB::Connection*) @ 0x100a3f89 in /usr/bin/clickhouse
4. DB::HedgedConnections::getReadyReplicaLocation(std::__1::function<void (int, Poco::Timespan const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&)>) @ 0x100a243f in /usr/bin/clickhouse
5. DB::HedgedConnections::receivePacketUnlocked(std::__1::function<void (int, Poco::Timespan const&, std::__1::basic_string<char, std::__1::char_traits<char>, std::__1::allocator<char> > const&)>) @ 0x100a323e in /usr/bin/clickhouse
6. DB::RemoteQueryExecutorRoutine::operator()(boost::context::fiber&&) const @ 0xf11cb52 in /usr/bin/clickhouse
7. void boost::context::detail::fiber_entry<boost::context::detail::fiber_record<boost::context::fiber, FiberStack&, DB::RemoteQueryExecutorRoutine> >(boost::context::detail::transfer_t) @ 0xf11ca2e in /usr/bin/clickhouse
 (version 21.4.4.30 (official build))



return res;
            default:
                
                disconnect();
                throw Exception("Unknown packet "
                    + toString(res.type)
                    + " from server " + getDescription(), ErrorCodes::UNKNOWN_PACKET_FROM_SERVER);
        }
    }
    catch (Exception & e)
    {
        
        e.setRemoteException();
        
        if (e.code() != ErrorCodes::UNKNOWN_PACKET_FROM_SERVER)
            e.addMessage("while receiving packet from " + getDescription());
        throw;
    }
}
Block Connection::receiveData()
{
    initBlockInput();
    return receiveDataImpl(block_in);
}
Block Connection::receiveLogData()
{
    initBlockLogsInput();
    return receiveDataImpl(block_logs_in);
}
Block Connection::receiveDataImpl(BlockInputStreamPtr & stream)