**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)
```
{
            LOG_WARNING(log,
                "Parallel distributed INSERT SELECT is not possible "
                "(source cluster={} ({} addresses), destination cluster={} ({} addresses))",
                storage_src ? storage_src->getClusterName() : "<not a Distributed table>",
                src_addresses.size(),
                getClusterName(),
                dst_addresses.size());
        }
        return nullptr;
    }
    if (settings.parallel_distributed_insert_select == PARALLEL_DISTRIBUTED_INSERT_SELECT_ALL)
    {
        new_query->table_id = StorageID(getRemoteDatabaseName(), getRemoteTableName());
        
        new_query->table_function.reset();
    }
    const auto & cluster = getCluster();
    const auto & shards_info = cluster->getShardsInfo();
    std::vector<std::unique_ptr<QueryPipelineBuilder>> pipelines;
    String new_query_str;
    {
        WriteBufferFromOwnString buf;
        IAST::FormatSettings ast_format_settings(buf,  true);
        ast_format_settings.always_quote_identifiers = true;
        new_query->IAST::format(ast_format_settings);
        new_query_str = buf.str();
    }
    ContextMutablePtr query_context = Context::createCopy(local_context);
    ++query_context->getClientInfo().distributed_depth;