```
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. 
```
{
                    if (job.stream)
                    {
                        pool->scheduleOrThrowOnError([&job]()
                        {
                            job.stream->writeSuffix();
                        });
                    }
                }
            }
        }
        catch (...)
        {
            pool->wait();
            throw;
        }
        try
        {
            pool->wait();
            log_performance();
        }
        catch (Exception & exception)
        {
            log_performance();
            exception.addMessage(getCurrentStateDescription());
            throw;
        }
    }
}
IColumn::Selector DistributedBlockOutputStream::createSelector(const Block & source_block) const