**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)
```
main_table = StorageID{remote_database, remote_table};
    ClusterProxy::SelectStreamFactory select_stream_factory =
        ClusterProxy::SelectStreamFactory(
            header,
            processed_stage);
    ClusterProxy::executeQuery(
        query_plan, header, processed_stage,
        main_table, remote_table_function_ptr,
        select_stream_factory, log, modified_query_ast,
        local_context, query_info,
        sharding_key_expr, sharding_key_column_name,
        query_info.cluster);
    
    if (!query_plan.isInitialized())
        throw Exception("Pipeline is not initialized", ErrorCodes::LOGICAL_ERROR);
}
SinkToStoragePtr StorageDistributed::write(const ASTPtr &, const StorageMetadataPtr & metadata_snapshot, ContextPtr local_context)
{
    auto cluster = getCluster();
    const auto & settings = local_context->getSettingsRef();
    
    if (!storage_policy && !owned_cluster && !settings.insert_distributed_sync && !settings.insert_shard_id)
    {
        throw Exception("Storage " + getName() + " must have own data directory to enable asynchronous inserts",
                        ErrorCodes::BAD_ARGUMENTS);
    }
    auto shard_num = cluster->getLocalShardCount() + cluster->getRemoteShardCount();
    
    if (!settings.insert_shard_id && !settings.insert_distributed_one_random_shard && !has_sharding_key && shard_num >= 2)
    {