```
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. 
```
job.connection_entry = connection_pool->get(timeouts, &settings);
                    if (job.connection_entry.isNull())
                        throw Exception("Got empty connection for replica" + replica.readableString(), ErrorCodes::LOGICAL_ERROR);
                }
                if (throttler)
                    job.connection_entry->setThrottler(throttler);
                job.stream = std::make_shared<RemoteBlockOutputStream>(*job.connection_entry, timeouts, query_string, settings, context.getClientInfo());
                job.stream->writePrefix();
            }
            CurrentMetrics::Increment metric_increment{CurrentMetrics::DistributedSend};
            job.stream->write(shard_block);
        }
        else 
        {
            if (!job.stream)
            {
                
                job.local_context = std::make_unique<Context>(context);
                InterpreterInsertQuery interp(query_ast, *job.local_context);
                auto block_io = interp.execute();
                job.stream = block_io.out;
                job.stream->writePrefix();
            }
            writeBlockConvert(job.stream, shard_block, shard_info.getLocalNodeCount());
        }
        job.blocks_written += 1;
        job.rows_written += shard_block.rows();
    };
}
void DistributedBlockOutputStream::writeSync(const Block & block)