**Use case**

Run view on each shard individually and insert into local node.

**Describe the solution you'd like**

```sql

INSERT INTO FUNCTION cluster('all-sharded', map, output, key_2) SELECT *
FROM cluster('all-sharded', view(
    SELECT *
    FROM map.input
    WHERE (key_1 % 2) = toUInt32(getMacro('all-sharded-shard'))
))
SETTINGS parallel_distributed_insert_select = 2

Query id: 0ef3707f-2b74-4367-bf6d-82939a64e0f2


0 rows in set. Elapsed: 0.002 sec.

Received exception from server (version 21.11.10):
Code: 60. DB::Exception: Received from localhost:9000. DB::Exception: Both table name and UUID are empty. (UNKNOWN_TABLE)
```
throw Exception(
            "Method write is not supported by storage " + getName() + " with more than one shard and no sharding key provided",
            ErrorCodes::STORAGE_REQUIRES_PARAMETER);
    }
    if (settings.insert_shard_id && settings.insert_shard_id > shard_num)
    {
        throw Exception("Shard id should be range from 1 to shard number", ErrorCodes::INVALID_SHARD_ID);
    }
    
    bool insert_sync = settings.insert_distributed_sync || settings.insert_shard_id || owned_cluster;
    auto timeout = settings.insert_distributed_timeout;
    Names columns_to_send;
    if (settings.insert_allow_materialized_columns)
        columns_to_send = metadata_snapshot->getSampleBlock().getNames();
    else
        columns_to_send = metadata_snapshot->getSampleBlockNonMaterialized().getNames();
    
    return std::make_shared<DistributedSink>(
        local_context, *this, metadata_snapshot, cluster, insert_sync, timeout,
        StorageID{remote_database, remote_table}, columns_to_send);
}
QueryPipelineBuilderPtr StorageDistributed::distributedWrite(const ASTInsertQuery & query, ContextPtr local_context)
{
    const Settings & settings = local_context->getSettingsRef();
    if (settings.max_distributed_depth && local_context->getClientInfo().distributed_depth >= settings.max_distributed_depth)
        throw Exception("Maximum distributed depth exceeded", ErrorCodes::TOO_LARGE_DISTRIBUTED_DEPTH);
    std::shared_ptr<StorageDistributed> storage_src;
    auto & select = query.select->as<ASTSelectWithUnionQuery &>();
    auto new_query = std::dynamic_pointer_cast<ASTInsertQuery>(query.clone());
    if (select.list_of_selects->children.size() == 1)