```
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. 
```
compression_level = settings.network_zstd_compression_level;
    CompressionCodecFactory::instance().validateCodec(compression_method, compression_level, !settings.allow_suspicious_codecs);
    CompressionCodecPtr compression_codec = CompressionCodecFactory::instance().get(compression_method, compression_level);
    
    
    std::string first_file_tmp_path{};
    auto reservation = storage.getStoragePolicy()->reserveAndCheck(block.bytes());
    const auto disk = reservation->getDisk();
    auto disk_path = disk->getPath();
    auto data_path = storage.getRelativeDataPath();
    auto make_directory_sync_guard = [&](const std::string & current_path)
    {
        std::unique_ptr<DirectorySyncGuard> guard;
        if (dir_fsync)
        {
            const std::string relative_path(data_path + current_path);
            guard = std::make_unique<DirectorySyncGuard>(disk, relative_path);
        }
        return guard;
    };
    auto it = dir_names.begin();
    
    
    {
        const std::string path(disk_path + data_path + *it);
        Poco::File(path).createDirectory();
        const std::string tmp_path(path + "/tmp/");
        Poco::File(tmp_path).createDirectory();
        const std::string file_name(toString(storage.file_names_increment.get()) + ".bin");
        first_file_tmp_path = tmp_path + file_name;