```
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. 
```
else
    {
        if (shard_info.isLocal() && settings.prefer_localhost_replica)
            writeToLocal(block, shard_info.getLocalNodeCount());
        std::vector<std::string> dir_names;
        for (const auto & address : cluster->getShardsAddresses()[shard_id])
            if (!address.is_local || !settings.prefer_localhost_replica)
                dir_names.push_back(address.toFullString(settings.use_compact_format_in_distributed_parts_names));
        if (!dir_names.empty())
            writeToShard(block, dir_names);
    }
}
void DistributedBlockOutputStream::writeToLocal(const Block & block, const size_t repeats)
{
    
    InterpreterInsertQuery interp(query_ast, context);
    auto block_io = interp.execute();
    block_io.out->writePrefix();
    writeBlockConvert(block_io.out, block, repeats);
    block_io.out->writeSuffix();
}
void DistributedBlockOutputStream::writeToShard(const Block & block, const std::vector<std::string> & dir_names)
{
    const auto & settings = context.getSettingsRef();
    const auto & distributed_settings = storage.getDistributedSettingsRef();
    bool fsync = distributed_settings.fsync_after_insert;
    bool dir_fsync = distributed_settings.fsync_directories;
    std::string compression_method = Poco::toUpper(settings.network_compression_method.toString());
    std::optional<int> compression_level;
    if (compression_method == "ZSTD")