```
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. 
```
{
    auto thread_group = CurrentThread::getGroup();
    return [this, thread_group, &job, &current_block, num_shards]()
    {
        if (thread_group)
            CurrentThread::attachToIfDetached(thread_group);
        setThreadName("DistrOutStrProc");
        ++job.blocks_started;
        SCOPE_EXIT({
            ++finished_jobs_count;
            UInt64 elapsed_time_for_block_ms = watch_current_block.elapsedMilliseconds();
            job.elapsed_time_ms += elapsed_time_for_block_ms;
            job.max_elapsed_time_for_block_ms = std::max(job.max_elapsed_time_for_block_ms, elapsed_time_for_block_ms);
        });
        const auto & shard_info = cluster->getShardsInfo()[job.shard_index];
        auto & shard_job = per_shard_jobs[job.shard_index];
        const auto & addresses = cluster->getShardsAddresses();
        
        if (num_shards > 1)
        {
            auto & shard_permutation = shard_job.shard_current_block_permutation;
            size_t num_shard_rows = shard_permutation.size();
            for (size_t j = 0; j < current_block.columns(); ++j)
            {
                const auto & src_column = current_block.getByPosition(j).column;
                auto & dst_column = job.current_shard_block.getByPosition(j).column;
                
                if (num_shard_rows)
                    dst_column = src_column->permute(shard_permutation, num_shard_rows);
                else