```
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. 
```
{
    return metadata_snapshot->getSampleBlock();
}
void DistributedBlockOutputStream::writePrefix()
{
}
void DistributedBlockOutputStream::write(const Block & block)
{
    Block ordinary_block{ block };
    /* They are added by the AddingDefaultBlockOutputStream, and we will get
     * different number of columns eventually */
    for (const auto & col : metadata_snapshot->getColumns().getMaterialized())
    {
        if (ordinary_block.has(col.name))
        {
            ordinary_block.erase(col.name);
            LOG_DEBUG(log, "{}: column {} will be removed, because it is MATERIALIZED",
                storage.getStorageID().getNameForLogs(), col.name);
        }
    }
    if (insert_sync)
        writeSync(ordinary_block);
    else
        writeAsync(ordinary_block);
}
void DistributedBlockOutputStream::writeAsync(const Block & block)