```
CREATE TABLE default.x AS system.numbers ENGINE = Log;

INSERT INTO FUNCTION cluster('test_cluster_two_shards', default, x) SELECT * FROM numbers(10)

Received exception from server (version 20.11.1):
Code: 55. DB::Exception: Received from localhost:9000. DB::Exception: Method write is not supported by storage Distributed with more than one shard and no sharding key provided. 

0 rows in set. Elapsed: 0.001 sec. 

INSERT INTO FUNCTION cluster('test_cluster_two_shards', default, x, rand()) SELECT * FROM numbers(10)

Received exception from server (version 20.11.1):
Code: 42. DB::Exception: Received from localhost:9000. DB::Exception: Table function 'cluster' requires from 2 to 3 parameters: <addresses pattern or cluster name>, <name of remote database>, <name of remote table>. 

0 rows in set. Elapsed: 0.001 sec. 
```
{
    Blocks splitted_blocks = splitBlock(block);
    const size_t num_shards = splitted_blocks.size();
    for (size_t shard_idx = 0; shard_idx < num_shards; ++shard_idx)
        if (splitted_blocks[shard_idx].rows())
            writeAsyncImpl(splitted_blocks[shard_idx], shard_idx);
    ++inserted_blocks;
}
void DistributedBlockOutputStream::writeAsyncImpl(const Block & block, const size_t shard_id)
{
    const auto & shard_info = cluster->getShardsInfo()[shard_id];
    const auto & settings = context.getSettingsRef();
    if (shard_info.hasInternalReplication())
    {
        if (shard_info.isLocal() && settings.prefer_localhost_replica)
            
            writeToLocal(block, shard_info.getLocalNodeCount());
        else
        {
            const auto & path = shard_info.insertPathForInternalReplication(
                settings.prefer_localhost_replica,
                settings.use_compact_format_in_distributed_parts_names);
            if (path.empty())
                throw Exception("Directory name for async inserts is empty", ErrorCodes::LOGICAL_ERROR);
            writeToShard(block, {path});
        }
    }